[#2191] Ingestion continue multi-machine → store unifié (la réplication EST le backup)

[myia-ai-01]

Parent : #2191

Driver — catastrophe 2026-05-28

Désinstallation de Roo Code fleet-wide (suspectée VS Code Settings Sync) → globalStorage effacés sans remontée préalable. Si l'ingestion continue multi-machine avait existé, aucune conversation n'aurait été perdue. La remontée régulière vers la DB unique EST le backup (combinée aux sources survivantes + GDrive).

Scope

Remontée régulière des tâches Roo / Zoo / Claude des 6 machines + Hermes/NanoClaw vers le store unifié :

• Postgres : DB unique sur ai-01, écriture distante depuis chaque machine via le sous-domaine dédié (pg.myia.io, SSL natif). Pas de réplication (décision user Q1, [#2191] Phase 2 — Prototype store unifié Postgres + Qdrant (schéma + writer + recherche 2-temps) #2426).
• Qdrant : indexation incrémentale (réutilise le leader-election indexeur feat(indexing): leader-election / partition task-space pour l'indexeur Qdrant (élimine la redondance N×) #2352, déjà livré, −96% GPU).
• Perf/latence : tenir compte de la lenteur rclone/FUSE pour les archives GDrive (cf write storm [REFONTE][CRITICAL] Mécanisme presence + config + sharedState GDrive — consolidation et fiabilisation #2121). Throttle + batch côté écriture distante.

Critères d'acceptation

• Chaque machine remonte ses tâches vers la DB unique avec latence bornée (cible < quelques min)
• ai-01 voit le squelette unifié cross-cluster via conversation_browser
• Perte/wipe d'une machine ≠ perte de données (tâches dans la DB + ré-ingestables depuis GDrive)
• Pas de write storm (mesure vs [REFONTE][CRITICAL] Mécanisme presence + config + sharedState GDrive — consolidation et fiabilisation #2121)

Cross-ref

#2121 (write storm GDrive), #2352 (indexeur leader-election — livré), #1747-D (validation cross-machine), #2426 (DB unique).

[myia-ai-01]

[CLAIMED] by claude on myia-ai-01 at 2026-05-29T23:07:59.3833982+02:00

[myia-ai-01]

[RESULT] myia-ai-01: PASS — completed (no code changes needed)

[jsboige]

Ingestion Mapping Design — #2427 Spec

Author: myia-po-2024 (claude-interactive, GLM-5.1)
Date: 2026-05-30

Current Data Flow (Pre-#2426)

Roo JSONL  -> RooStorageDetector  -> SkeletonCacheService -> Qdrant (embeddings)
Claude JSONL -> ClaudeStorageDetector -/                            |
GDrive archives -> TaskArchiver -> archiveToSkeleton() -> conversation_browser

3 tiers in SkeletonCacheService:

• Tier 1: Roo local (tasks/ dir, ui_messages.json)
• Tier 2: Claude Code local (~/.claude/projects/ JSONL)
• Tier 3: GDrive cross-machine archives ($ROOSYNC_SHARED_PATH/task-archive/{machineId}/)

Key types:

• ArchivedTask (v2): { taskId, machineId, hostIdentifier, archivedAt, metadata: ArchivedTaskMetadata, messages: ArchivedTaskMessage[] }
• ConversationSkeleton: { taskId, parentTaskId, isCompleted, metadata: SkeletonMetadata, sequence: (MessageSkeleton|ActionMetadata)[] }
• SkeletonHeader: { taskId, parentTaskId, metadata: SkeletonMetadata, childTaskInstructionPrefixes? }

Target Data Flow (Post-#2426)

Roo JSONL  -> SkeletonCacheService.addOrUpdate() -> local skeleton cache -> Qdrant
                    | (dual-write hook, best-effort)                       |
                    v                                                    v
              UnifiedStoreWriter.upsert() -> Postgres (ai-01 unique)

Ingestion Mapping: ArchivedTask -> Postgres Schema

Per #2426 spec: conversations PK task_id + messages PK id + tool_calls JSONB FK.

conversations table

Postgres column	Source (ArchivedTask / ConversationSkeleton)	Transform
task_id (PK)	archive.taskId	Direct
parent_task_id	archive.metadata.parentTaskId or skeleton.parentTaskId	Nullable, camelCase
machine_id	archive.machineId	Direct
host_identifier	archive.hostIdentifier	Direct
title	archive.metadata.title	Nullable
workspace	archive.metadata.workspace	Nullable
mode	archive.metadata.mode	Nullable
source	archive.metadata.source	"roo" / "claude-code" / "zoo-code"
is_completed	archive.metadata.isCompleted	Boolean
message_count	archive.metadata.messageCount	Integer
first_ts	archive.messages[0].timestamp or archive.metadata.createdAt	MUST use message-level timestamp, NOT mtime (anti-catastrophe rule from spec)
last_ts	archive.messages[last].timestamp or archive.metadata.lastActivity	Same rule
created_at	archive.metadata.createdAt	Nullable
last_activity	archive.metadata.lastActivity	Nullable
archived_at	archive.archivedAt	ISO 8601
ingested_at	now()	Server-side default
data_source	"archive" / "local"	Distinguishes ingestion path

messages table

Postgres column	Source	Transform
id (PK)	Generated: {taskId}_{idx}	Deterministic, enables idempotent re-ingestion
task_id (FK)	archive.taskId	References conversations.task_id
role	message.role	"user" / "assistant"
content	message.content	Full text (v2 archives: no truncation)
timestamp	message.timestamp or fallback archive.archivedAt	Same anti-catastrophe rule
sequence_index	Array index in archive.messages	0-based, preserves ordering
tool_calls	NULL for archive ingestion	Populated only from live JSONL parsing (Claude Code tool_use blocks)

Ingestion Paths

Path 1: Live dual-write (SkeletonCacheService.addOrUpdate hook)

addOrUpdate(taskId, skeleton)
  -> existing: saveSkeleton() to disk
  -> NEW: UnifiedStoreWriter.upsert(skeleton) // best-effort, fire-and-forget

• Trigger: Every skeleton add/update during normal operation
• Data source: ConversationSkeleton (full sequence available)
• Tool calls: Extracted from ActionMetadata entries in sequence
• Throttle: ~30s batch (Q7 from spec), not per-call
• Error handling: Non-blocking (catch + warn, like existing GDrive archive write)

Path 2: Batch backfill from GDrive archives (293 archives)

TaskArchiver.listArchivedTasks() -> for each:
  TaskArchiver.readArchivedTask(taskId) -> ArchivedTask
    -> UnifiedStoreWriter.upsertArchive(archive) // new method

• Trigger: Manual rebuild_cache or new ingest_archives tool
• Data source: ArchivedTask (v2 preferred, v1 tolerated)
• Tool calls: NULL (archives don't preserve tool_call structure, only message content)
• Batch size: ~50 per batch with 30s throttle between batches
• Idempotent: upsert on conflict (deterministic PKs)

Path 3: Claude Code JSONL direct (for tool_calls enrichment)

ClaudeStorageDetector.detectClaudeConversations()
  -> parse JSONL -> extract tool_use blocks -> messages with tool_calls JSONB

• Trigger: Live ingestion only (not backfill)
• Data source: Raw JSONL from ~/.claude/projects/
• Tool calls: Full extraction from type: "tool_use" / type: "tool_result" blocks
• This is Phase B/C of [#2191] Phase 2 — Prototype store unifié Postgres + Qdrant (schéma + writer + recherche 2-temps) #2426 — not part of initial ingestion

Migration: Archive -> Postgres

-- Ingested from ArchivedTask v2
INSERT INTO conversations (task_id, parent_task_id, machine_id, title, workspace,
    mode, source, is_completed, message_count, first_ts, last_ts, data_source)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 'archive')
ON CONFLICT (task_id) DO UPDATE SET
    message_count = EXCLUDED.message_count,
    last_ts = EXCLUDED.last_ts,
    data_source = EXCLUDED.data_source;

-- Messages: batch insert with deterministic IDs
INSERT INTO messages (id, task_id, role, content, timestamp, sequence_index)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (id) DO UPDATE SET
    content = EXCLUDED.content,
    timestamp = EXCLUDED.timestamp;

Write Storm Mitigation (#2121)

• Throttle: 30s batch window (Q7 spec). Accumulate upserts in a local buffer, flush every 30s.
• Backpressure: If Postgres write fails (connection timeout), pause buffer flush for 60s, retry with exponential backoff (1/2/4 min).
• Archives backfill: Process in batches of 50 with 30s pause between batches.
• GDrive FUSE: NOT involved in Postgres writes (Postgres is direct TCP to pg.myia.io). Only reading archives FROM GDrive for backfill.

Dependency on #2426

This spec is design-only until #2426 Phase B (writer implementation) lands from po-2023. When it does:

1. UnifiedStoreWriter.upsertArchive(archive: ArchivedTask) — new method for Path 2
2. UnifiedStoreWriter.upsert(skeleton: ConversationSkeleton) — existing method, extended for Path 1
3. New tool: ingest_archives — triggers Path 2 backfill

Cross-ref

• [RESPEC #1393] conversation_browser sur stockage GDrive partagé + index/DB distribuée (online-only, multi-harnais) #2191 (Epic), [#2191] Phase 2 — Prototype store unifié Postgres + Qdrant (schéma + writer + recherche 2-temps) #2426 (Postgres prototype), [#2191] Re-injection des archives récupérées (catastrophe 2026-05-28) dans le store unifié #2428 (re-injection archives), [REFONTE][CRITICAL] Mécanisme presence + config + sharedState GDrive — consolidation et fiabilisation #2121 (write storm), feat(indexing): leader-election / partition task-space pour l'indexeur Qdrant (élimine la redondance N×) #2352 (leader-election indexeur)