diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b1712f..bc5177c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ All notable changes are documented here. +## Unreleased + +- Detect and quarantine abort events that are semantically closed but incompatible with Codex desktop projection. +- Repair interrupted turns with Codex-native completion metadata while preserving an explicit non-fabricated tool-output placeholder. +- Publish the current synchronization result in the same device report instead of reporting the previous run. + ## 0.1.0 - 2026-07-12 - Initial private repository baseline. diff --git a/SKILL.md b/SKILL.md index 2caa97a..2ea9eba 100644 --- a/SKILL.md +++ b/SKILL.md @@ -65,7 +65,7 @@ Manage project folders through the background CLI/API; do not open the Syncthing ## Safety behavior - Treat conversation JSONL as append-only. Publish a per-device head and advance the canonical copy only when all heads have a prefix relationship. -- Require semantic completeness in addition to valid JSON: every tool call needs a matching output, and every older turn needs `task_complete` or `turn_aborted`. +- Require semantic and desktop-projection completeness in addition to valid JSON: every tool call needs a matching output, every older turn needs `task_complete` or `turn_aborted`, and repaired aborts need Codex-compatible completion fields. - Never export the active turn or the tool call currently running Sync2. When possible, publish the semantically complete prefix before the active turn as a stable checkpoint; otherwise skip the task until a healthy checkpoint exists. - Quarantine incomplete local snapshots, heads, and canonical files from import or promotion while continuing independent skill/project work. - On divergent conversation history, preserve every head, emit a conflict, and do not invent a merged history. @@ -82,6 +82,6 @@ Manage project folders through the background CLI/API; do not open the Syncthing Read [usage.md](references/usage.md) for daily commands and output interpretation. For first-machine, new-machine, reinstall, or path-change deployment, follow [deployment.md](references/deployment.md). For edge cases, conflict semantics, and storage boundaries, consult [protocol.md](references/protocol.md). -After setup or material changes, run `conversation audit `, `doctor`, `sync --dry-run`, `sync`, `device report`, and `status`. A healthy active task may report one open latest turn and active calls; it must report zero persistent dangling calls and zero stale open turns. Outside maintenance, `doctor` must also require a stable canonical and stable head from every device. +After setup or material changes, run `conversation audit `, `doctor`, `sync --dry-run`, `sync`, `device report`, and `status`. A healthy active task may report one open latest turn and active calls; it must report zero persistent dangling calls, zero stale open turns, and zero projection-unsafe abort events. Outside maintenance, `doctor` must also require a stable canonical and stable head from every device. If an imported task does not appear immediately in the Codex sidebar, verify its JSONL and `state_5.sqlite` row with `doctor`, then restart Codex once. Do not copy the source device's whole database. diff --git a/references/protocol.md b/references/protocol.md index 75025e1..6288e82 100644 --- a/references/protocol.md +++ b/references/protocol.md @@ -46,7 +46,7 @@ Never synchronize local config or base state between devices. They contain devic Treat Codex rollout JSONL as append-only. Read only complete JSON records; discard an incomplete final line from a live writer snapshot. -Before publishing, validate both syntax and event semantics. Pair every `custom_tool_call`/`function_call` with its output and close every older turn with `task_complete` or `turn_aborted`. Treat calls belonging to the latest open turn as transient. Never publish that active turn; if the prefix before its `task_started` event is independently stable, publish that completed prefix as a checkpoint. This prevents long-running Goals from starving sync when a new turn starts immediately after each completion. +Before publishing, validate syntax, event semantics, and desktop projection compatibility. Pair every `custom_tool_call`/`function_call` with its output and close every older turn with `task_complete` or `turn_aborted`. A repaired abort must match Codex's native projection shape: an associated `` developer marker followed by `reason: "interrupted"`, numeric `completed_at`, and numeric `duration_ms`. Treat calls belonging to the latest open turn as transient. Never publish that active turn; if the prefix before its `task_started` event is independently stable, publish that completed prefix as a checkpoint. This prevents long-running Goals from starving sync when a new turn starts immediately after each completion. Each device publishes its stable local snapshot to `heads/.jsonl`. Reconciliation compares raw bytes: @@ -54,7 +54,7 @@ Each device publishes its stable local snapshot to `heads/.jsonl`. Recon - when any pair diverges, retain the existing canonical, preserve all heads, and write `conflict.json`; - never interleave turns or invent a merged event order. -Ignore semantically incomplete heads and canonical files. Continue skill and project synchronization instead of failing the entire run. Once a valid longer head arrives, it can replace an invalid old canonical without treating the invalid copies as merge candidates. +Ignore semantically incomplete or projection-unsafe heads and canonical files. Continue skill and project synchronization instead of failing the entire run. Once a valid longer head arrives, it can replace an invalid old canonical without treating the invalid copies as merge candidates. Import only a semantically complete canonical into the platform-appropriate `~/.codex/sessions/YYYY/MM/DD` path. Update that thread in `state_5.sqlite`, `session_index.jsonl`, and an existing local desktop catalog row. Do not copy database files between devices. @@ -88,7 +88,7 @@ Case-only file or directory names are unsafe across Windows and default macOS fi ## 5. Device reports and cold start -`device report` publishes content-stable evidence about platform, Node version, protocol revision, Sync2 script SHA-256, vault, selected-task local/vault health, skill counts, scheduler state, and last-run result. `device list` reads all propagated reports. Do not declare a fleet upgraded until every expected device reports the same current protocol revision and script hash. +`device report` publishes content-stable evidence about platform, Node version, protocol revision, Sync2 script SHA-256, vault, selected-task local/vault health, skill counts, scheduler state, and last-run result. A normal sync embeds that same run's result in the report before transport commit; if the transport commit fails, it rewrites the local/vault report as failed instead of leaving a false success. `device list` reads all propagated reports. Do not declare a fleet upgraded until every expected device reports the same current protocol revision and script hash. On a new device, the first sync usually performs this order: @@ -166,8 +166,8 @@ Resolution archives nonchosen vault heads and backs up a differing local history 1. Disable the scheduler and pause the transport folder. 2. Back up the rollout, state database, session index, desktop catalog database, and vault task directory. 3. Run `conversation audit `. -4. Run `conversation repair --title `. The repair appends an explicit interrupted-output placeholder and aborts only stale older turns; it does not fabricate the original output or close the currently executing turn. -5. Require zero persistent dangling calls, zero stale open turns, and consistent titles across indexes. +4. Run `conversation repair --title `. The repair appends an explicit interrupted-output placeholder and aborts only stale older turns using Codex's native projection-compatible event shape; it does not fabricate the original output or close the currently executing turn. +5. Require zero persistent dangling calls, zero stale open turns, zero projection-unsafe abort events, and consistent titles across indexes. 6. Leave invalid old heads quarantined until a stable local final turn publishes a valid replacement canonical. ### Skill file diff --git a/scripts/sync2.mjs b/scripts/sync2.mjs index a7a0e67..9e38322 100644 --- a/scripts/sync2.mjs +++ b/scripts/sync2.mjs @@ -177,6 +177,7 @@ function auditJsonlBuffer(buffer) { const startedTurns = []; const startedTurnEvents = []; const closedTurns = new Set(); + const latestTurnAbortEvents = new Map(); let finalAnswers = 0; let userMessages = 0; let lastTimestamp = null; @@ -187,9 +188,23 @@ function auditJsonlBuffer(buffer) { const turnId = payload.turn_id ?? payload.internal_chat_message_metadata_passthrough?.turn_id ?? null; if (event.type === "event_msg" && payload.type === "task_started" && payload.turn_id) { startedTurns.push(payload.turn_id); - startedTurnEvents.push({ turnId: payload.turn_id, line: index + 1 }); + startedTurnEvents.push({ turnId: payload.turn_id, line: index + 1, startedAt: payload.started_at ?? null }); + } + if (event.type === "event_msg" && ["task_complete", "turn_aborted"].includes(payload.type) && payload.turn_id) { + closedTurns.add(payload.turn_id); + if (payload.type === "turn_aborted") { + latestTurnAbortEvents.set(payload.turn_id, { + turnId: payload.turn_id, + line: index + 1, + reason: payload.reason ?? null, + completedAt: payload.completed_at ?? null, + durationMs: payload.duration_ms ?? null, + projectionCompatible: payload.reason === "interrupted" + && Number.isFinite(payload.completed_at) + && Number.isFinite(payload.duration_ms), + }); + } } - if (event.type === "event_msg" && ["task_complete", "turn_aborted"].includes(payload.type) && payload.turn_id) closedTurns.add(payload.turn_id); if ((payload.type === "custom_tool_call" || payload.type === "function_call") && payload.call_id) { calls.set(payload.call_id, { callId: payload.call_id, kind: payload.type, line: index + 1, turnId }); } @@ -204,6 +219,7 @@ function auditJsonlBuffer(buffer) { const activeDanglingCalls = danglingCalls.filter((item) => latestTurnId && item.turnId === latestTurnId && openTurns.includes(latestTurnId)); const tailDanglingCalls = danglingCalls.filter((item) => item.line === lines.length); const persistentDanglingCalls = danglingCalls.filter((item) => !activeDanglingCalls.some((active) => active.callId === item.callId)); + const projectionUnsafeAbortEvents = [...latestTurnAbortEvents.values()].filter((item) => !item.projectionCompatible); return { lines: lines.length, bytes: buffer.length, @@ -216,11 +232,12 @@ function auditJsonlBuffer(buffer) { activeDanglingCalls, tailDanglingCalls, persistentDanglingCalls, + projectionUnsafeAbortEvents, openTurns, staleOpenTurns, latestTurnId, - semanticOk: persistentDanglingCalls.length === 0 && staleOpenTurns.length === 0, - stable: danglingCalls.length === 0 && openTurns.length === 0, + semanticOk: persistentDanglingCalls.length === 0 && staleOpenTurns.length === 0 && projectionUnsafeAbortEvents.length === 0, + stable: danglingCalls.length === 0 && openTurns.length === 0 && projectionUnsafeAbortEvents.length === 0, }; } @@ -1202,7 +1219,7 @@ function reconcileProjectShares(config, dryRun, summary) { } } -function buildDeviceReport(config) { +function buildDeviceReport(config, lastRunOverride = undefined) { const selectedConversations = Object.entries(config.conversations ?? {}).filter(([, event]) => event.selected).map(([id, event]) => { const rollout = findRollout(config.codexHome, id); let audit = null; @@ -1240,12 +1257,12 @@ function buildDeviceReport(config) { skillSources, daemon: daemonStatus(config, config._configFile), maintenance: { enabled: fs.existsSync(maintenanceFile(config)) }, - lastRun: readJson(lastRunFile(config), null), + lastRun: lastRunOverride === undefined ? readJson(lastRunFile(config), null) : lastRunOverride, }; } -function publishDeviceReport(config, dryRun = false) { - const report = buildDeviceReport(config); +function publishDeviceReport(config, dryRun = false, lastRunOverride = undefined) { + const report = buildDeviceReport(config, lastRunOverride); writeJson(path.join(config.vault, ".sync2", "device-reports", `${config.deviceId}.json`), report, dryRun); return report; } @@ -1290,6 +1307,7 @@ function syncCommand(options, direction = "sync") { const dryRun = Boolean(options["dry-run"]); const summary = newSummary(direction, dryRun); const startedAt = nowIso(); + let reportPublished = false; try { ensureVault(config, dryRun); if (!dryRun && fs.existsSync(maintenanceFile(config)) && !options.force) { @@ -1300,14 +1318,21 @@ function syncCommand(options, direction = "sync") { syncConversations(config, direction, dryRun, summary); syncSkills(config, direction, dryRun, summary); reconcileProjectShares(config, dryRun, summary); - publishDeviceReport(config, dryRun); + const successfulRun = { ok: true, startedAt, finishedAt: nowIso(), summary }; + recordLastRun(config, successfulRun, dryRun); + publishDeviceReport(config, dryRun, successfulRun); + reportPublished = !dryRun; saveConfig(file, config, dryRun); }); gitPost(config, dryRun); - recordLastRun(config, { ok: true, startedAt, finishedAt: nowIso(), summary }, dryRun); return summary; } catch (error) { - recordLastRun(config, { ok: false, startedAt, finishedAt: nowIso(), error: error.message, direction }, dryRun); + const failedRun = { ok: false, startedAt, finishedAt: nowIso(), error: error.message, direction }; + recordLastRun(config, failedRun, dryRun); + if (reportPublished) { + try { withVaultLock(config, false, () => publishDeviceReport(config, false, failedRun)); } + catch { /* Preserve the original sync failure. The next successful run republishes health. */ } + } throw error; } } @@ -1410,8 +1435,39 @@ function repairConversationCommand(options, selector) { }); } } - for (const turnId of before.rollout.staleOpenTurns) { - additions.push({ timestamp: nowIso(), type: "event_msg", payload: { type: "turn_aborted", turn_id: turnId, reason: "recovered_stale_turn" } }); + const turnsNeedingProjectionRepair = new Set([ + ...before.rollout.staleOpenTurns, + ...before.rollout.projectionUnsafeAbortEvents.map((item) => item.turnId), + ]); + for (const turnId of turnsNeedingProjectionRepair) { + const completedAt = Math.floor(Date.now() / 1000); + const startedAt = before.rollout.startedTurnEvents.find((item) => item.turnId === turnId)?.startedAt; + const durationMs = Number.isFinite(startedAt) ? Math.max(0, (completedAt - startedAt) * 1000) : 0; + additions.push({ + timestamp: nowIso(), + type: "response_item", + payload: { + type: "message", + role: "developer", + content: [{ + type: "input_text", + text: "\nThe previous turn was interrupted before its completion record was persisted. Sync2 recovered the stale turn without reconstructing missing execution output.\n", + }], + internal_chat_message_metadata_passthrough: { turn_id: turnId }, + }, + }); + additions.push({ + timestamp: nowIso(), + type: "event_msg", + payload: { + type: "turn_aborted", + turn_id: turnId, + reason: "interrupted", + recovery_reason: "recovered_stale_turn", + completed_at: completedAt, + duration_ms: durationMs, + }, + }); } if (additions.length) fs.appendFileSync(rollout, `${additions.map((item) => JSON.stringify(item)).join("\n")}\n`); const state = threadRow(config.codexHome, before.id) ?? {}; @@ -1444,9 +1500,11 @@ function repairConversationCommand(options, selector) { id: before.id, recovery, repairedCalls: before.rollout.persistentDanglingCalls.map((item) => item.callId), - closedStaleTurns: before.rollout.staleOpenTurns, + closedStaleTurns: [...turnsNeedingProjectionRepair], title, - semanticOkAfter: after.rollout.persistentDanglingCalls.filter((item) => !before.rollout.tailDanglingCalls.some((tail) => tail.callId === item.callId)).length === 0 && after.rollout.staleOpenTurns.length === 0, + semanticOkAfter: after.rollout.persistentDanglingCalls.filter((item) => !before.rollout.tailDanglingCalls.some((tail) => tail.callId === item.callId)).length === 0 + && after.rollout.staleOpenTurns.length === 0 + && after.rollout.projectionUnsafeAbortEvents.length === 0, indexesConsistent: after.indexes.consistent, }; } diff --git a/scripts/test-sync2.mjs b/scripts/test-sync2.mjs index 82293c8..2658f66 100644 --- a/scripts/test-sync2.mjs +++ b/scripts/test-sync2.mjs @@ -149,6 +149,9 @@ try { assert.equal(reportB.selectedConversations[0].rolloutExists, true); assert.equal(reportB.selectedConversations[0].indexedInStateDb, true); assert.ok(reportB.skillSources[0].files >= 1); + assert.equal(reportB.lastRun.ok, true); + assert.equal(reportB.lastRun.summary.conversationsPulled, pullB.conversationsPulled); + assert.equal(reportB.lastRun.summary.skillFilesPulled, pullB.skillFilesPulled); const dbB = new DatabaseSync(path.join(b.home, "state_5.sqlite")); assert.equal(dbB.prepare("SELECT title FROM threads WHERE id=?").get(id).title, "Sync2 E2E Thread"); dbB.close(); @@ -195,8 +198,12 @@ try { const preview = run(["sync", "--dry-run", "--config", configA]); assert.equal(preview.dryRun, true); - const daemonPreview = run(["daemon", "install", "--dry-run", "--minutes", "3", "--config", configA]); - assert.equal(daemonPreview.action, "install-preview"); + if (["win32", "darwin"].includes(process.platform)) { + const daemonPreview = run(["daemon", "install", "--dry-run", "--minutes", "3", "--config", configA]); + assert.equal(daemonPreview.action, "install-preview"); + } else { + assert.match(runFails(["daemon", "install", "--dry-run", "--minutes", "3", "--config", configA]), /supports Windows Task Scheduler and macOS LaunchAgents/); + } const doctor = run(["doctor", "--config", configA]); assert.equal(doctor.ok, true); const status = run(["status", "--config", configA]); @@ -254,11 +261,34 @@ try { const repairedCanonical = fs.readFileSync(path.join(damagedConversationRoot, "canonical.jsonl"), "utf8"); assert.match(repairedCanonical, /custom_tool_call_output/); assert.match(repairedCanonical, /recovered_stale_turn/); + const repairedAudit = run(["conversation", "audit", damagedId, "--config", damagedConfigA]); + assert.equal(repairedAudit.rollout.projectionUnsafeAbortEvents.length, 0); + const repairedAbort = repairedCanonical.split(/\r?\n/).filter(Boolean).map((line) => JSON.parse(line)) + .find((item) => item.type === "event_msg" && item.payload?.type === "turn_aborted" && item.payload?.turn_id === oldTurn); + assert.equal(repairedAbort.payload.reason, "interrupted"); + assert.equal(Number.isFinite(repairedAbort.payload.completed_at), true); + assert.equal(Number.isFinite(repairedAbort.payload.duration_ms), true); assert.ok(fs.existsSync(path.join(damagedConversationRoot, "metadata", "win-c.json"))); const repairedPull = run(["pull", "--config", damagedConfigB]); assert.equal(repairedPull.conversationsPulled, 1); assert.ok(findRollout(damagedB.home, damagedId)); + const legacyAbortTurn = "turn-legacy-projection-abort"; + fs.appendFileSync(damagedRollout, `${record("event_msg", { type: "task_started", turn_id: legacyAbortTurn, started_at: 1767225900 })}\n${record("event_msg", { type: "turn_aborted", turn_id: legacyAbortTurn, reason: "recovered_stale_turn" })}\n`); + const legacyAbortAudit = run(["conversation", "audit", damagedId, "--config", damagedConfigA]); + assert.equal(legacyAbortAudit.rollout.semanticOk, false); + assert.equal(legacyAbortAudit.rollout.projectionUnsafeAbortEvents.length, 1); + const legacyAbortRepair = run(["conversation", "repair", damagedId, "--config", damagedConfigA]); + assert.deepEqual(legacyAbortRepair.closedStaleTurns, [legacyAbortTurn]); + assert.equal(legacyAbortRepair.semanticOkAfter, true); + const legacyAbortAfter = run(["conversation", "audit", damagedId, "--config", damagedConfigA]); + assert.equal(legacyAbortAfter.rollout.projectionUnsafeAbortEvents.length, 0); + const legacyAbortEvents = fs.readFileSync(damagedRollout, "utf8").split(/\r?\n/).filter(Boolean).map((line) => JSON.parse(line)) + .filter((item) => item.type === "event_msg" && item.payload?.type === "turn_aborted" && item.payload?.turn_id === legacyAbortTurn); + assert.equal(legacyAbortEvents.at(-1).payload.reason, "interrupted"); + assert.equal(Number.isFinite(legacyAbortEvents.at(-1).payload.completed_at), true); + assert.equal(Number.isFinite(legacyAbortEvents.at(-1).payload.duration_ms), true); + run(["maintenance", "on", "--reason", "test", "--config", damagedConfigA]); assert.match(runFails(["sync", "--config", damagedConfigA]), /maintenance mode is enabled/); const maintenanceStatus = run(["maintenance", "status", "--config", damagedConfigA]); @@ -307,7 +337,7 @@ try { assert.equal(relocation.action, "vault-updated"); assert.ok(fs.existsSync(path.join(relocatedVault, "skills", "codex", "git-skill", "SKILL.md"))); - console.log(JSON.stringify({ ok: true, root, checks: 56, deviceReports: true, noOpWrites: true, stableActiveCheckpoint: true, extendedWindowsPath: true, semanticRepair: true, unsafeCanonicalQuarantine: true, activeTurnProtection: true, perDeviceMetadata: true, desktopCatalogImport: true, maintenanceMode: true, maintenanceBootstrap: process.platform === "win32", folderTransport: true, gitTransport: true, vaultRelocation: true, conversationImport: true, skillThreeWay: true, conflictRecovery: true, daemonPreview: true }, null, 2)); + console.log(JSON.stringify({ ok: true, root, checks: 65, deviceReports: true, noOpWrites: true, stableActiveCheckpoint: true, extendedWindowsPath: true, semanticRepair: true, legacyAbortRepair: true, unsafeCanonicalQuarantine: true, activeTurnProtection: true, perDeviceMetadata: true, desktopCatalogImport: true, maintenanceMode: true, maintenanceBootstrap: process.platform === "win32", folderTransport: true, gitTransport: true, vaultRelocation: true, conversationImport: true, skillThreeWay: true, conflictRecovery: true, daemonPreview: true }, null, 2)); } catch (error) { console.error(`E2E FAILED; artifacts kept at ${root}`); throw error;