--- name: conversation-archive description: Source-agnostic ingest for conversation transcripts. One skill for WhatsApp `_chat.txt`, Telegram, Signal, LinkedIn DMs, Zoom transcript, meeting minutes, iMessage, Slack, X DMs, email threads, and any future source — `source` is a property on the parent `:ConversationArchive`, never a separate skill or plugin. Confirms the owner + every distinct sender against existing `:AdminUser` / `:Person` nodes (no auto-creation), then invokes the deterministic Bash entry `conversation-archive-ingest.sh` with `--source `. The script normalises the source (per-source pluggable function), sessionizes at gap-hours boundaries, and emits one JSON line containing the prepared sessions array (turn-attributed text per session + per-session cursor). The dispatched specialist then produces a typed-section JSON chunking for each session in-turn and calls the `memory-ingest` MCP tool with that session's sections + cursor — :Section chunks land under a parent `:ConversationArchive` MERGEd on `conversationIdentity = sha256(accountId + ":" + sortedParticipantElementIds + ":" + source)` atomically with the cursor advance, so a kill mid-archive resumes from the next session on re-issue. Re-imports are delta-append: prior chunks are never touched unless `--rebuild` is set. Triggers when the operator drops any conversation export (file or directory) into chat. --- # Conversation Archive — source-agnostic transcript ingestion **Invoked from `specialists:librarian`** — the admin agent dispatches the librarian when the operator drops a conversation transcript; the librarian loads this skill. **Default parent.** Every `:ConversationArchive` (previously a `:KnowledgeDocument` carrying `conversationIdentity`) is anchored to the operator-confirmed owner (`:AdminUser` for the account owner, `:Person` for a third-party archive). The owner's elementId is derived inside the bash entry from `(ACCOUNT_ID, USER_ID)` env (see Identity below) and returned in the prepare JSON; it becomes the `anchorNodeId` argument to `memory-ingest` for every session of the run. Per-session `:Section` chunks attach to the archive via `:HAS_SECTION`, chained in reading order by `:NEXT`. Participant cross-edges from the archive to each distinct sender (`:Person` / `:AdminUser`) attach via `:PARTICIPANT_IN` — they describe the conversation, they do not substitute for the parent. One skill, one bash entry, one writer. The pipeline is identical for every conversation source. The only source-specific code is the per-source normaliser (a pluggable function under [`platform/plugins/memory/mcp/src/lib/conversation-normalisers/`](../../mcp/src/lib/conversation-normalisers/)). New sources = a new normaliser file + an enum entry, never a new skill. ## Where the work lives The deterministic work — parsing, sessionising, archiveSha256/conversationIdentity, prior-cursor lookup, delta slicing — lives in the bash entry and runs once per ingest. The LLM-judgment work — producing topic-bounded `:Section` chunks with summary + keywords for each session — lives in the dispatched specialist's per-turn work, one session at a time, against the turn-attributed text the prepare script returns. The specialist calls the `memory-ingest` MCP tool once per session with that session's classified chunks + the session's per-session cursor; the writer commits chunks + cursor advance in one Cypher transaction, so a kill mid-archive resumes from the next session on re-issue without re-classifying anything already written. ## Confirmed sources (Phase 0) `whatsapp` ships today (relocated grammar from the retired `whatsapp-import` plugin). `x-dm` ships alongside the `x-import` plugin — driven by the x-import skill, one invocation per dmConversation slice. Other sources land as needed: - `telegram` — Telegram JSON export - `signal` — Signal text export - `linkedin-messages` — LinkedIn DM thread export (LinkedIn *connections* stay flat-dataset; only DMs route here) - `zoom-transcript` — Zoom .vtt / .txt transcript - `meeting-minutes` — operator-supplied meeting notes - `imessage` — iMessage backup export - `slack` — Slack channel export - `x-dm` — X (Twitter) Direct Messages, dispatched by the `x-import` skill from `data/direct-messages.js` / `data/direct-messages-group.js` — one bash invocation per dmConversation slice; senders carry numeric X senderIds resolved to `:Person` / `:AdminUser` upstream in the x-import skill - `email` — RFC 5322 email thread, dispatched by the `email` plugin's writer. Participants resolve to existing `:Person` / `:AdminUser` by the `email` property on the canonical node — closed-set, no auto-creation, same discipline as LinkedIn DMs. Until a normaliser ships, the bash entry loud-fails with the implemented set listed in the error. ## Step 1 — owner + all-participants confirmation (mandatory first step) A conversation transcript carries N distinct senders (1 owner + N-1 others). Every distinct senderName must resolve to an existing `:AdminUser` or `:Person` elementId before the script runs — the writer LOUD-FAILs on any unresolved sender. 1. **Identify the source.** Look at the file the operator dropped — `_chat.txt` ⇒ `whatsapp`; `*.json` from a Telegram export ⇒ `telegram`; `.vtt` ⇒ `zoom-transcript`; `.txt` of formatted meeting notes ⇒ `meeting-minutes`; etc. If unsure, ask the operator one question to disambiguate. 2. **Read the file head** to discover senders before the script runs. Read the first ~50 lines (or use a per-source preview heuristic). Extract distinct sender display names. Surface a one-line histogram (`Sender '' ( messages)`) per distinct sender. 3. **Iterate the histogram, one operator question per distinct senderName.** For each sender: `"Sender '' ( messages) — pick existing :AdminUser/:Person or block?"`. The operator either picks an existing elementId or names "block" (refuses to map to a node). **Never auto-create a `:Person`** — the operator must confirm a canonical node. 4. **Identify the owner** from the resolved set — the operator who exported the chat. Echo back: `"Owner = :AdminUser (); other participants = . Confirm yes/no."` 5. **Persist the resolved IDs** as `--participant-person-ids ` for the script call. The owner's elementId is not passed as a flag — the script derives it from `(ACCOUNT_ID, USER_ID)` env. DM and group follow the identical flow. A 1:1 chat resolves 2 senders; a group resolves N senders. Identity (`conversationIdentity = sha256(accountId + ":" + sortedParticipantElementIds.join(",") + ":" + source)`) is the same regardless of group size for a given source. Source is part of the hash: "me + Joe on WhatsApp" and "me + Joe by email" yield two distinct archive parents, not one merged row — without source-scoping their `:Person` elementIds match and the rows collide. ## Step 2 — prepare (one synchronous bash call) `ACCOUNT_ID` and `USER_ID` come from the agent's process env (plumbed by `spawn-env.ts` into every Bash subprocess). The script LOUD-FAILs on missing or malformed values; the operator never passes them as flags. The owner `:AdminUser` elementId is derived from `(ACCOUNT_ID, USER_ID)` inside the script via Cypher. Invoke the entry **synchronously** (no `run_in_background`): ```bash bash platform/plugins/memory/bin/conversation-archive-ingest.sh \ --source \ --participant-person-ids ,,... \ --scope ``` Optional flags: - `--timezone ` — IANA zone for timestamps (default `Europe/London`). - `--date-format
` — WhatsApp only; override auto-detect for ambiguous locales. - `--rebuild` — operator-issued only; never agent-autonomous. Treats the run as first-ingest, signals that the first session's `memory-ingest` call must carry `cleanupPriorChunks=true` to drop prior chunks for this `archiveSha256`. Sessions split deterministically at 8h gap (fixed code constant). The legacy `--session-gap-hours` flag is removed; passing it FAILs at `phase=argv`. The script exits 0 with one JSON line on stdout (see the shape below) or exits non-zero with one stderr line: `[conversation-archive] FAIL phase= reason="..."`. Surface FAIL verbatim and yield — never retry. ### Prepare-output JSON shape ```json { "archiveElementId": "4:abcd…:42" /* or null on first-ingest */, "conversationIdentity": "", "archiveSha256": "", "archiveSourceFile": "_chat.txt", "archiveTitle": "whatsapp · Joel ↔ Adam · 2024-01-15→2026-04-30", "source": "whatsapp", "scope": "admin", "accountId": "", "ownerElementId": "", "participantElementIds": ["", "..."] /* non-owner participants */, "rebuild": false, "parsed": 1707, "mediaSkipped": 0, "systemSkipped": 0, "delta": { "kind": "first-ingest|delta|rebuild|empty-delta", "deltaStart": 0, "deltaMessages": 1707 }, "dateRange": { "first": "2024-01-15T09:30:00+00:00", "last": "2026-04-30T18:42:00+01:00" }, "senderHistogram": [{ "name": "Joel", "count": 812 }, { "name": "Adam", "count": 895 }], "priorChunkCount": 0, "priorLastIngestedMessageAt": null, "sessions": [ { "sessionIndex": 1, "totalSessions": 38, "messageCount": 42, "firstMessageAt": "2024-01-15T09:30:00+00:00", "lastMessageAt": "2024-01-15T17:42:00+00:00", "lastMessageHash": "", "turnText": "[2024-01-15T09:30:00+00:00] Joel: Morning\n[2024-01-15T09:31:00+00:00] Adam: hey\n...", "turnTextPath": "/tmp/maxy-conversation-archive/-/session-1.txt" }, "..." ], "ms": 6800 } ``` **Empty-delta short-circuit.** When `delta.kind === "empty-delta"`, the `sessions` array is empty and `priorChunkCount` carries the existing chunk count. No `memory-ingest` calls are needed; jump to the operator messages section with a noop summary. **`turnTextPath` is diagnostic only.** The bin writes each session's `turnText` to a deterministic tempfile at that path; the `memory-ingest` writer reconstructs the same path from `(conversationIdentity, archiveSha256, sessionIndex)` and reads it directly. The specialist does not pass `turnTextPath` to `memory-ingest`. The field is emitted so an operator can `ls /tmp/maxy-conversation-archive/` when debugging a stuck ingest. ## Step 3 — classify and write, one session per turn For each `session` in the prepare JSON, in order: 1. **Read** the session's `turnText` (turn-attributed `[ISO timestamp] Sender: body\n…`). 2. **Produce a typed-section JSON chunking in-turn.** Split the turn text into topic-bounded `:Section` chunks following the standard ontology shape: each section carries `kind` (`Conversation` for a generic chunk, or a typed kind like `Topic` when the conversation explicitly establishes a typed entity), `title`, `summary` (1-3 sentences, ≤500 chars), `sourceStart` / `sourceEnd` (char offsets into this session's `turnText`), `properties` (carry `firstMessageAt`, `lastMessageAt`, and `messageCount` from the session — or finer per-chunk slices when the chunking subdivides), and `anchorEdge: null` (chunks attach via the parent's `:HAS_SECTION` chain). **Do NOT emit `body`.** The writer slices `:Section.body` server-side from the bin's per-session tempfile using the `(sourceStart, sourceEnd)` offsets — this closes the same LLM-paraphrase vector closed on the document path. Any `body` you emit is logged and ignored. Also produce a `documentKeywords` array of 3-8 topic keywords for the session, deduplicated across the archive. 3. **Call the `memory-ingest` MCP tool** with this session's payload (conversation-archive path): - `conversationIdentity` ← from the prepare JSON. - `source` ← from the prepare JSON. - `archiveSha256` ← from the prepare JSON. - `archiveSourceFile` ← from the prepare JSON. - `archiveTitle` ← from the prepare JSON. - `sessionIndex` ← `session.sessionIndex` from the prepare JSON. **Required** — the writer uses it together with `conversationIdentity` and `archiveSha256` to locate the per-session turnText tempfile and slice `:Section.body` server-side. Missing or non-integer values LOUD-FAIL. - `participantElementIds` ← `[ownerElementId, ...participantElementIds]` from the prepare JSON (owner first). - `anchorNodeId` ← `ownerElementId` from the prepare JSON. - `anchorLabel` ← `"AdminUser"`. - `documentSummary` ← `archiveTitle` (stable label). - `sections` ← the typed-section JSON you produced (offsets only, no `body`). - `documentKeywords` ← the topic keywords for this session. - `lastIngestedMessageHash` ← `session.lastMessageHash`. - `lastIngestedMessageAt` ← `session.lastMessageAt`. - `scope` ← from the prepare JSON. - `cleanupPriorChunks` ← `true` ONLY when `prepare.rebuild === true && session.sessionIndex === 1`. False on every other call. 4. **Track running totals** (`totalChunksWritten`, `totalNextEdges`, `participantsLinked` on the first session, `cleanedPriorChunks` on the first session, the union of `documentKeywords`). The MCP response carries `sectionCount`, `edgeBreakdown.NEXT`, `edgeBreakdown.PARTICIPANT_IN`, and `cleanedPriorChunks` for each call; aggregate them. 5. **If `memory-ingest` returns an error**, stop immediately. The cursor for sessions already committed is durable on the `:ConversationArchive` node — re-issuing the entire bash + classify loop with the same archive will resume from the next un-committed session. The classify-then-write per-session pattern is the resumption-safety primitive: each session's chunks + cursor advance happen in one Cypher transaction inside `memory-ingest`. A kill mid-archive (operator cancel, SDK silence-timeout, network blip) leaves the cursor at session N-1's last message; the operator re-runs the same skill and only session N onward classifies. ## Step 4 — three operator messages per ingest After the last session lands, formulate the three operator-facing messages from the prepare JSON + the aggregated MCP responses: 1. **Parse summary.** `Parsed (source=): messages across sessions, date range → . Participants: .` 2. **Classify summary.** `Classified into chunks, covering: .` 3. **Write summary.** `Created :ConversationArchive with :Section chunks (NEXT chain length ). Participants linked via :PARTICIPANT_IN: .` For an empty-delta re-import (`delta.kind === "empty-delta"`): emit only message 1 + a noop line `noop reason="no new messages since "`. If the first session's response carries `cleanedPriorChunks > 0` and `prepare.rebuild === false`, surface a WARN line — destructive cleanup outside of an operator-issued `--rebuild` is a doctrine violation. ## Failure path — single FAIL line - **Exit non-zero on prepare** + one stderr line: `[conversation-archive] FAIL phase= reason="..."`. Surface verbatim and yield. Do not retry. - **`memory-ingest` error mid-loop**: stop and surface the error. The cursor for sessions already committed survives; a fresh skill invocation resumes from the next un-committed session. - `delta-cursor-missing` LOUD-FAIL: the prior `lastIngestedMessageHash` is not present in the re-export. Either the operator deleted prior messages, or this is a different chat. Investigation required — never re-run blindly. ## Idempotency - **Re-running the same export bytes is a no-op by default.** The script returns an empty-delta JSON with the existing chunk count; no `memory-ingest` calls run. - **Re-running with appended messages** (fresh export from the same chat with new messages at the tail): the script slices new messages, sessionises only those, and the specialist's per-session `memory-ingest` calls append new chunks at the tail of the existing `:NEXT` chain. Pre-existing chunks are never touched. - **Destructive rebuild** requires the explicit `--rebuild` flag — operator-issued only. With `--rebuild`, the prepare script treats the run as first-ingest; the specialist's first `memory-ingest` call must carry `cleanupPriorChunks: true` (subsequent calls must carry `false` or they would delete chunks just written). - **Re-ingesting into a previously-trashed archive** revives the parent before MERGE. `memory-ingest`'s `ingestConversationDocument` strips `:Trashed` from the matching `conversationIdentity` so the MERGE binds to a non-trashed parent. ## Doctrine - **`--rebuild` is operator-issued only.** The agent must NEVER propose `--rebuild` as a fix for any FAIL phase, classifier defect, or partial-state recovery. The operator types `--rebuild` themselves. - **Use a fresh export** (different `archiveSha256`) for every legitimate re-classify case the agent surfaces. Same-bytes re-classify is the operator's call, not the agent's. - **One `memory-ingest` call per session.** Never batch sessions into one call (it breaks the atomic per-session cursor) and never skip the per-session cursor fields (the writer fails without them on the conversation-archive path). ## Verification (post-write) - `MATCH (a:ConversationArchive { conversationIdentity: $cid }) RETURN elementId(a), a.source, a.lastIngestedMessageAt, a.lastIngestedMessageHash` — `source` matches `--source`; counters agree with the aggregated MCP responses. - `MATCH (a:ConversationArchive { conversationIdentity: $cid })-[:HAS_SECTION]->(c:Section) RETURN count(c)` — equals the aggregated `sectionCount`. - `MATCH (p)-[:PARTICIPANT_IN]->(:ConversationArchive { conversationIdentity: $cid }) RETURN count(p)` — equals `participantsLinked` after a first-ingest. - Phase 2 (`:Observation` / `:Task` / `:Preference` derivation) is its own follow-up. ## What this is not - **Not** the live `whatsapp` plugin. That plugin (Baileys QR pairing) holds messages in an in-memory store cleared on restart. This skill imports historical exports into Neo4j as persistent graph nodes. - **Not** a media-transcription pipeline. Voice notes, photos, PDFs are skipped at parse with a counter logged. - **Not** an insight-extraction pass. Phase 2 (`:Observation` / `:Task` / `:Preference` / `:MENTIONS` derivation, anchored to chunks) ships in its own task and applies uniformly to every source. - **Not** an archive-wide importer. Flat datasets (LinkedIn connections, CRM exports) are first-class entities + natural edges, not `:ConversationArchive` and not `:KnowledgeDocument`. - **Not** automatic. The owner + all-participants confirmation gate is mandatory before any line is written. - **Not** the primary path for Granola / Otter / Circleback meetings. Those three providers ship first-class MCP servers; the canonical ingest path is the sibling [`conversation-archive-mcp`](../conversation-archive-mcp/SKILL.md) skill, which shares this writer surface and identity formula. This file-drop path stays available as a fallback if Granola / Otter / Circleback ever ship per-provider export normalisers (none today; the brief is to use the MCP).