From 92422caec9760d12790772b1d9e447050e1f2b13 Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 15:05:48 -0500 Subject: [PATCH 1/8] fix: keep hot memory local while hosted work waits --- .github/workflows/coding-hosts.yml | 50 +++++++ .gitignore | 2 + docs/HOT-SYNC.md | 26 +++- package-lock.json | 4 +- package.json | 2 +- scripts/profile-hot-sync.mjs | 73 ++++++++++ scripts/smoke-coding-hosts.mjs | 131 +++++++++++++++++ src/runtime/hot-sync-manager.ts | 72 ++++++++- src/runtime/lifecycle-engine.ts | 115 +++++++++++++-- tests/runtime/hot-sync-acceptance.test.ts | 64 ++++++++ tests/runtime/hot-sync-hook-path.test.ts | 169 ++++++++++++++++++++++ 11 files changed, 680 insertions(+), 28 deletions(-) create mode 100644 .github/workflows/coding-hosts.yml create mode 100644 scripts/profile-hot-sync.mjs create mode 100644 scripts/smoke-coding-hosts.mjs create mode 100644 tests/runtime/hot-sync-hook-path.test.ts diff --git a/.github/workflows/coding-hosts.yml b/.github/workflows/coding-hosts.yml new file mode 100644 index 0000000..2c50e91 --- /dev/null +++ b/.github/workflows/coding-hosts.yml @@ -0,0 +1,50 @@ +name: Coding host customer journey + +on: + workflow_dispatch: + pull_request: + paths: + - 'src/runtime/**' + - 'scripts/smoke-coding-hosts.mjs' + - '.github/workflows/coding-hosts.yml' + - 'package-lock.json' + +permissions: + contents: read + +concurrency: + group: coding-hosts-${{ github.ref }} + cancel-in-progress: true + +jobs: + customer-journey: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest] + runs-on: ${{ matrix.os }} + timeout-minutes: 20 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 # v7.1.0 + with: + node-version: 22 + cache: npm + - run: npm ci + - run: npm run release:build + - name: Install pinned real coding hosts + run: npm install --prefix "$RUNNER_TEMP/coding-hosts" --ignore-scripts --no-audit --no-fund @openai/codex@0.164.0 @anthropic-ai/claude-code@2.1.319 + - name: Exercise clean candidate install and real host hooks + env: + CROSSHEAD_TEST_CLAUDE_BIN: ${{ runner.temp }}/coding-hosts/node_modules/.bin/claude + CROSSHEAD_TEST_CODEX_BIN: ${{ runner.temp }}/coding-hosts/node_modules/.bin/codex + run: node scripts/smoke-coding-hosts.mjs + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + if: always() + with: + name: coding-host-journey-${{ matrix.os }} + path: .coding-host-test-logs/ + include-hidden-files: true + retention-days: 14 diff --git a/.gitignore b/.gitignore index d67ef55..553604b 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,5 @@ release/ .native-host-test-logs/ *.tgz .DS_Store + +.coding-host-test-logs/ diff --git a/docs/HOT-SYNC.md b/docs/HOT-SYNC.md index 7c8d132..6834642 100644 --- a/docs/HOT-SYNC.md +++ b/docs/HOT-SYNC.md @@ -1,16 +1,20 @@ # Local hot memory with background sync -The `feat/local-first-hot-memory` branch provides the opt-in implementation of -hosted Task 159 / PRD 007. Enable `CROSSHEAD_HOT_SYNC_ENABLED=true` in the daemon's -environment, with a server running migration 0034 and the same flag. Start with -a disposable runtime home. This alpha candidate is not a published release. +Local hot sync is available in the v0.5.0 alpha series. Enable +`CROSSHEAD_HOT_SYNC_ENABLED=true` in the daemon's environment, with a server +running migration 0034 and the same flag. All participating devices must use +protocol 2. Production capacity validation is required before changing the +launch default or advertising multi-device recovery as generally available. ## What happens locally A completed lifecycle capture is signed and written atomically, with file and directory fsync, before it is acknowledged. Hooks read the verified local view; ordinary hot-context delivery does not wait for authentication or the cloud. -Explicit paid recall remains a hosted operation. Local-only profiles retain +Explicit paid recall remains a hosted operation. Ordinary hooks do not request +billable context packs or reflection, and do not wait for hosted inbox/dream HUD +refreshes. Archive uploads run outside the lifecycle mutation lock and cannot +block local capture or hot sync. Local-only profiles retain their existing behavior. Pending checkpoints have a separate durable outbox. They survive restart, @@ -75,6 +79,15 @@ continuity. Legacy writers use a separate endpoint and cannot overwrite the revisioned store. Unmapped remote workspaces remain inspectable until linked locally and never enter an unrelated workspace's prompt. +Unsent legacy queue checkpoints are imported into the fsynced outbox before +new uploads. Their exact original snapshots remain in private +`legacy-recovery-*.json` files, listed by `legacyRecoverySnapshots` in sync +status. These bounded recovery copies protect ambiguous work if first discovery +reveals a reset on another machine. Unsupported snapshots and snapshots with a +known reset remain queued for explicit recovery; they do not block other +profiles' local hooks. Do not automatically re-upload recovery copies after a +reset without deciding whether they predate the deletion. + Deploy the server capability before releasing the client. If the server gate is disabled or the server is rolled back, leave the client gate enabled: local capture/read continues, pending uploads remain durable, and status reports the @@ -90,8 +103,7 @@ Retry-After; and conflict withholding. The compiled daemon test uses SIGKILL between capture and read, then verifies Codex, Claude Code, and Pi context through its authenticated local API, including reset and ownership controls. -This branch still requires public release and deployment validation before a -launch: tagged artifacts, production installer/update/rollback checks, a real +Each launch candidate requires public release and deployment validation: tagged artifacts, production installer/update/rollback checks, a real two-device production run, and authenticated production capacity measurements. Local test latency is development evidence, not a production capacity claim. diff --git a/package-lock.json b/package-lock.json index cd90e06..9aa9064 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@crosshead/runtime", - "version": "0.5.0-alpha.6", + "version": "0.5.0-alpha.7", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@crosshead/runtime", - "version": "0.5.0-alpha.6", + "version": "0.5.0-alpha.7", "license": "Apache-2.0", "dependencies": { "@noble/curves": "^1.9.7", diff --git a/package.json b/package.json index ac69ffd..55f8b8e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@crosshead/runtime", - "version": "0.5.0-alpha.6", + "version": "0.5.0-alpha.7", "description": "Crosshead local runtime: persistent memory for Claude Code, Codex, Pi, Hermes and OpenCode. CLI, daemon, host adapters and hooks.", "license": "Apache-2.0", "private": true, diff --git a/scripts/profile-hot-sync.mjs b/scripts/profile-hot-sync.mjs new file mode 100644 index 0000000..62cc042 --- /dev/null +++ b/scripts/profile-hot-sync.mjs @@ -0,0 +1,73 @@ +#!/usr/bin/env node +// Bounded workload using the shipped manager, disposable identities and synthetic findings. +import { createRequire } from 'node:module'; +import { mkdtempSync, rmSync } from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { performance } from 'node:perf_hooks'; +const require = createRequire(import.meta.url); +const { generateSecretKey, getPublicKey, nip19 } = require('nostr-tools'); +const { HotSyncManager } = require('../dist/runtime/hot-sync-manager.js'); +const { HostedOnlyBotsClient } = require('../dist/runtime/hosted-client.js'); +const option = (name, fallback) => { const i = process.argv.indexOf(name); return i < 0 ? fallback : process.argv[i + 1]; }; +const origin = new URL(option('--base-url', 'https://crosshead.sh')).origin; +if (!process.argv.includes('--allow-production') && !['localhost', '127.0.0.1'].includes(new URL(origin).hostname)) throw new Error('Remote profiling requires --allow-production'); +const count = Number(option('--clients', '6')), seconds = Number(option('--seconds', '65')), captureMs = Number(option('--capture-ms', '3000')); +if (!Number.isSafeInteger(count) || count < 1 || count > 10 || seconds < 10 || seconds > 180 || captureMs < 1500) throw new Error('Budget: 1–10 clients, 10–180 seconds, captures at least 1500 ms apart'); +const root = mkdtempSync(path.join(os.tmpdir(), 'crosshead-sync-profile-')); +const clients = [], managers = [], observations = [], captureLatencies = []; +const originalFetch = globalThis.fetch; +let abort = null, phase = 'workload'; +globalThis.fetch = async (...args) => { + const started = performance.now(); + const response = await originalFetch(...args); + const route = new URL(String(args[0])).pathname; + observations.push({ phase, route, method: args[1]?.method ?? 'GET', status: response.status, ms: performance.now() - started }); + if (phase === 'workload' && (response.status === 429 || response.status >= 500)) abort ??= { route, status: response.status, retryAfter: response.headers.get('retry-after') }; + return response; +}; +const sleep = ms => new Promise(resolve => setTimeout(resolve, ms)); +const quantile = (a, q) => [...a].sort((x,y) => x-y)[Math.max(0, Math.ceil(a.length*q)-1)] ?? 0; +const summary = { origin, clients: count, seconds, captureMs, startedAt: new Date().toISOString(), synthetic: true, cleanup: [] }; +try { + for (let i = 0; i < count; i++) { + const key = generateSecretKey(), identity = { nsec: nip19.nsecEncode(key), npub: nip19.npubEncode(getPublicKey(key)) }; + const client = new HostedOnlyBotsClient({ apiBase: origin, ...identity }); + clients.push(client); managers.push(new HotSyncManager({ stateDir: path.join(root, String(i)), origin, ...identity, profileName: 'launch-capacity-fixture' }, client)); + } + const start = performance.now(); let nextCapture = 0, event = 0; + while (!abort && performance.now() - start < seconds*1000) { + if (performance.now() - start >= nextCapture) { + for (const [i, manager] of managers.entries()) { + const before = performance.now(); + manager.capture('codex', { id: `capacity-${i}-${event}`, runtime: 'codex', workspaceId: 'repo:/synthetic-crosshead-capacity', occurredAt: new Date().toISOString(), goal: 'Synthetic launch sync verification', summary: `Synthetic capture ${event}; no customer content.` }); + captureLatencies.push(performance.now()-before); + } + event++; nextCapture += captureMs; + } + for (const manager of managers) manager.tick(); + await sleep(100); + } + summary.elapsedSeconds = Number(((performance.now()-start)/1000).toFixed(2)); + summary.abort = abort; + summary.captures = captureLatencies.length; + summary.captureP95Ms = Number(quantile(captureLatencies,.95).toFixed(2)); + summary.devices = managers.map(m => { const s = m.status(); return { state: s.state, pending: s.pendingCount, error: s.error }; }); +} finally { + phase = 'cleanup'; + await Promise.all(managers.map(m => m.stop())); + for (const [i, manager] of managers.entries()) { + const client = clients[i]; + for (const {scope} of manager.status().scopes) { + try { const base = await client.readHotSync(manager.namespace, scope); await client.resetHotSync(base); summary.cleanup.push({ device: i, scopeCleared: true }); } + catch (error) { summary.cleanup.push({ device: i, scopeCleared: false, status: error.status, code: error.code }); } + } + const session = client.session(); + if (session) await originalFetch(`${origin}/v1/auth/revoke`, { method: 'POST', headers: { authorization: `Bearer ${session.sessionToken}` }, signal: AbortSignal.timeout(10000) }); + } + summary.requests = Object.values(observations.reduce((all, x) => { const key = `${x.phase} ${x.method} ${x.route}`; const row = all[key] ??= { route: key, count: 0, statuses: {}, latencies: [] }; row.count++; row.statuses[x.status] = (row.statuses[x.status] ?? 0)+1; row.latencies.push(x.ms); return all; }, {})).map(({latencies, ...row}) => ({ ...row, p95Ms: Math.round(quantile(latencies,.95)) })); + summary.finishedAt = new Date().toISOString(); + globalThis.fetch = originalFetch; rmSync(root, { recursive: true, force: true }); + console.log(JSON.stringify(summary, null, 2)); +} +if (summary.abort || summary.cleanup.some(x => !x.scopeCleared)) process.exitCode = 2; diff --git a/scripts/smoke-coding-hosts.mjs b/scripts/smoke-coding-hosts.mjs new file mode 100644 index 0000000..1a257c2 --- /dev/null +++ b/scripts/smoke-coding-hosts.mjs @@ -0,0 +1,131 @@ +#!/usr/bin/env node +// Real Claude/Codex processes, a deterministic local model, generated identity. +// This proves hook delivery and durability; it does not measure model quality. +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { createServer } from 'node:http'; +import { createRequire } from 'node:module'; +import { createHash } from 'node:crypto'; +import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, rmSync } from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +const require = createRequire(import.meta.url); +const tar = require('tar'); +const root = mkdtempSync(path.join(os.tmpdir(), 'crosshead-coding-hosts-')); +const output = path.resolve(process.env.CROSSHEAD_TEST_LOG_DIR ?? '.coding-host-test-logs'); +mkdirSync(output, { recursive: true }); +const installed = path.join(root, 'installed'), home = path.join(root, 'runtime'), project = path.join(root, 'project'); +const claudeHome = path.join(root, 'claude'), codexHome = path.join(root, 'codex'); +for (const dir of [installed, home, project, claudeHome, codexHome]) mkdirSync(dir, { recursive: true, mode: 0o700 }); +const nativeClaude = process.env.CROSSHEAD_TEST_CLAUDE_BIN; +const nativeCodex = process.env.CROSSHEAD_TEST_CODEX_BIN; +assert(nativeClaude && nativeCodex, 'Set CROSSHEAD_TEST_CLAUDE_BIN and CROSSHEAD_TEST_CODEX_BIN to the actual pinned host binaries'); +// Construct a fresh environment instead of inheriting provider/config credentials. +const env = { PATH: `${path.dirname(process.execPath)}:${process.env.PATH}`, TMPDIR: process.env.TMPDIR ?? os.tmpdir(), LANG: 'en_US.UTF-8', HOME: root, + XDG_CONFIG_HOME: path.join(root, 'config'), XDG_DATA_HOME: path.join(root, 'data'), XDG_CACHE_HOME: path.join(root, 'cache'), + CROSSHEAD_HOME: home, ONLYBOTS_HOME: home, CROSSHEAD_HOT_SYNC_ENABLED: 'true', + CLAUDE_CONFIG_DIR: claudeHome, CLAUDE_CODE_HOME: claudeHome, CODEX_HOME: codexHome, + CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: '1', DISABLE_AUTOUPDATER: '1', DISABLE_TELEMETRY: '1', + ANTHROPIC_API_KEY: 'fixture-only', ANTHROPIC_AUTH_TOKEN: 'fixture-only', FIXTURE_API_KEY: 'fixture-only' }; +let stage = 'capture', requests = [], cleanup = async () => {}; +const marker = 'duration-fixture-8427'; +const finding = `Finding ${marker}: serialize durations as decimal nanosecond strings. Zero is valid; negative durations are rejected. Do not convert to floating-point milliseconds. Next validate the transport round trip.`; +const server = createServer(async (req, res) => { + if (req.method === 'GET') { res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify({ object: 'list', data: [{ id: 'crosshead-fixture', object: 'model' }] })); return; } + let raw = ''; for await (const chunk of req) raw += chunk; + const body = JSON.parse(raw || '{}'); requests.push({ stage, path: req.url, body }); + if (req.url.includes('count_tokens')) { res.setHeader('Content-Type', 'application/json'); res.end('{"input_tokens":200}'); return; } + const text = stage === 'capture' ? finding : `Continue transport validation. Fixture stage ${stage}.`; + const emit = (event, data) => res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + if (req.url.includes('messages')) { + const message = { id: 'msg_fixture', type: 'message', role: 'assistant', model: body.model, content: [{ type: 'text', text }], stop_reason: 'end_turn', stop_sequence: null, usage: { input_tokens: 200, output_tokens: 60 } }; + if (!body.stream) { res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(message)); return; } + res.writeHead(200, { 'Content-Type': 'text/event-stream' }); + emit('message_start', { type: 'message_start', message: { ...message, content: [], stop_reason: null, usage: { input_tokens: 200, output_tokens: 0 } } }); + emit('content_block_start', { type: 'content_block_start', index: 0, content_block: { type: 'text', text: '' } }); + emit('content_block_delta', { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text } }); + emit('content_block_stop', { type: 'content_block_stop', index: 0 }); + emit('message_delta', { type: 'message_delta', delta: { stop_reason: 'end_turn', stop_sequence: null }, usage: { output_tokens: 60 } }); + emit('message_stop', { type: 'message_stop' }); res.end(); return; + } + const item = { id: 'msg_fixture', type: 'message', role: 'assistant', status: 'completed', content: [{ type: 'output_text', text, annotations: [] }] }; + const response = { id: 'resp_fixture', object: 'response', created_at: 1800000000, model: body.model, status: 'completed', output: [item], usage: { input_tokens: 200, output_tokens: 60, total_tokens: 260 } }; + if (!body.stream) { res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(response)); return; } + res.writeHead(200, { 'Content-Type': 'text/event-stream' }); + let sequence = 0; const responseEvent = (type, data) => emit(type, { type, sequence_number: sequence++, ...data }); + responseEvent('response.created', { response: { ...response, status: 'in_progress', output: [] } }); + responseEvent('response.output_item.added', { output_index: 0, item: { ...item, status: 'in_progress', content: [] } }); + responseEvent('response.content_part.added', { item_id: item.id, output_index: 0, content_index: 0, part: { type: 'output_text', text: '', annotations: [] } }); + responseEvent('response.output_text.delta', { item_id: item.id, output_index: 0, content_index: 0, delta: text }); + responseEvent('response.output_text.done', { item_id: item.id, output_index: 0, content_index: 0, text }); + responseEvent('response.content_part.done', { item_id: item.id, output_index: 0, content_index: 0, part: item.content[0] }); + responseEvent('response.output_item.done', { output_index: 0, item }); + responseEvent('response.completed', { response }); res.end(); +}); +await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); +const base = `http://127.0.0.1:${server.address().port}`; +env.ANTHROPIC_BASE_URL = base; +const run = async (name, binary, args, cwd = project) => { + const child = spawn(binary, args, { cwd, env, stdio: ['ignore', 'pipe', 'pipe'] }); + let stdout = '', stderr = ''; child.stdout.on('data', data => stdout += data); child.stderr.on('data', data => stderr += data); + const timer = setTimeout(() => child.kill('SIGKILL'), 90000); + const code = await new Promise((resolve, reject) => { child.on('error', reject); child.on('exit', resolve); }).finally(() => clearTimeout(timer)); + writeFileSync(path.join(output, `${name}.log`), `${stdout}\nSTDERR:\n${stderr}`); + assert.equal(code, 0, `${name} exited ${code}; inspect ${output}/${name}.log`); return stdout; +}; +const results = { platform: process.platform, checks: [], fixture: true }; +try { + const releaseDir = path.resolve(process.argv[2] ?? 'release'); + const manifest = JSON.parse(readFileSync(path.join(releaseDir, 'crosshead-release.json'), 'utf8')); + const archive = path.join(releaseDir, manifest.artifacts.install.file); + assert.equal(createHash('sha256').update(readFileSync(archive)).digest('hex'), manifest.artifacts.install.sha256); + await tar.x({ file: archive, cwd: installed }); + results.hosts = { claude: (await run('claude-version', nativeClaude, ['--version'])).trim(), codex: (await run('codex-version', nativeCodex, ['--version'])).trim() }; + await run('runtime-dependencies', 'npm', ['ci', '--omit=dev', '--ignore-scripts', '--no-audit', '--no-fund'], installed); + const fromInstalled = createRequire(path.join(installed, 'package.json')); + const { generateSecretKey, getPublicKey, nip19 } = fromInstalled('nostr-tools'); + const persistence = fromInstalled('./dist/runtime/runtime-persistence.js'); + const { LocalRuntimeClient } = fromInstalled('./dist/runtime/local-client.js'); + const { stopDetachedRuntimeDaemon } = fromInstalled('./dist/runtime/local-runtime.js'); + cleanup = () => stopDetachedRuntimeDaemon(home); + const key = generateSecretKey(), now = new Date().toISOString(), paths = persistence.ensureRuntimeHomeLayout(home); + persistence.saveProfile(paths, { name: 'default', apiBase: 'https://crosshead.invalid', npub: nip19.npubEncode(getPublicKey(key)), memoryMode: 'assist', workspaceScope: 'repo', features: { feedback: false, relay: false, coldSync: false, autoWarmPromotion: false }, secureStoreMode: 'file', createdAt: now, updatedAt: now }); + persistence.saveSecret(paths, 'default', { nsec: nip19.nsecEncode(key), session: null }); + writeFileSync(path.join(claudeHome, 'settings.json'), JSON.stringify({ env: { FIXTURE_PRESERVED: 'yes' } })); + writeFileSync(path.join(codexHome, 'config.toml'), `model = "crosshead-fixture"\nmodel_provider = "crosshead_fixture"\n[model_providers.crosshead_fixture]\nname = "Local deterministic fixture"\nbase_url = "${base}/v1"\nenv_key = "FIXTURE_API_KEY"\nwire_api = "responses"\nrequires_openai_auth = false\nsupports_websockets = false\nrequest_max_retries = 0\nstream_max_retries = 0\n`); + const cli = path.join(installed, 'dist/runtime/cli.js'); + const installClaude = ['install', 'claude-code', '--home', home, '--claude-home', claudeHome, '--claude-bin', nativeClaude, '--workspace-root', project]; + const installCodex = ['install', 'codex', '--home', home, '--codex-home', codexHome, '--codex-bin', nativeCodex]; + await run('install-claude', process.execPath, [cli, ...installClaude]); + await run('install-codex', process.execPath, [cli, ...installCodex]); + const client = new LocalRuntimeClient({ home, timeoutMs: 5000 }); + await run('claude-first', path.join(home, 'bin/claude'), ['-p', 'Investigate duration encoding and preserve the boundary decisions.', '--model', 'claude-fixture', '--tools', '', '--output-format', 'json']); + for (let i = 0; i < 100; i++) { + if (JSON.stringify(await client.readHotMemory('default', 'all')).includes(marker)) break; + if (i === 99) throw new Error('Claude final response was not captured'); + await new Promise(resolve => setTimeout(resolve, 100)); + } + assert(requests.some(r => r.stage === 'capture' && JSON.stringify(r.body).includes('Crosshead hooks automatically capture')), 'Claude did not load installed hooks'); + results.checks.push('real Claude host loaded installed hooks', 'assistant finding captured durably while hosted service unavailable'); + await cleanup(); stage = 'cross'; + await run('codex-fresh', path.join(home, 'bin/codex'), ['exec', '--skip-git-repo-check', '--dangerously-bypass-hook-trust', '--sandbox', 'read-only', '--json', 'Continue the duration investigation.']); + assert(requests.some(r => r.stage === 'cross' && JSON.stringify(r.body).includes(marker)), 'Fresh Codex did not receive Claude finding'); + results.checks.push('real fresh Codex receives Claude finding through installed hooks'); + await cleanup(); stage = 'fresh'; + await run('claude-fresh', path.join(home, 'bin/claude'), ['-p', 'Recall the duration encoding boundary decisions.', '--model', 'claude-fixture', '--tools', '', '--output-format', 'json']); + assert(requests.some(r => r.stage === 'fresh' && JSON.stringify(r.body).includes(marker)), 'Fresh Claude session did not receive prior finding'); + results.checks.push('fresh Claude session after daemon restart recalls finding'); + await run('adapter-reinstall', process.execPath, [cli, ...installClaude]); + await run('uninstall-claude', process.execPath, [cli, 'uninstall', 'claude-code', '--home', home, '--claude-home', claudeHome]); + await run('uninstall-codex', process.execPath, [cli, 'uninstall', 'codex', '--home', home, '--codex-home', codexHome]); + assert.equal(JSON.parse(readFileSync(path.join(claudeHome, 'settings.json'), 'utf8')).env.FIXTURE_PRESERVED, 'yes'); + assert(readFileSync(path.join(codexHome, 'config.toml'), 'utf8').includes('[model_providers.crosshead_fixture]')); + assert(!JSON.stringify(JSON.parse(readFileSync(path.join(claudeHome, 'settings.json'), 'utf8'))).includes('hook-runtime')); + results.checks.push('adapter reinstall and uninstall preserve unrelated host settings'); + results.releaseId = manifest.releaseId; results.source = manifest.source; + writeFileSync(path.join(output, 'result.json'), JSON.stringify(results, null, 2) + '\n'); + console.log(JSON.stringify(results)); +} finally { + writeFileSync(path.join(output, 'provider-observations.json'), JSON.stringify(requests.map(({stage, path: route, body}) => ({ stage, route, hasFinding: JSON.stringify(body).includes(marker), hasCaptureContract: JSON.stringify(body).includes('Crosshead hooks automatically capture') })), null, 2) + '\n'); + await cleanup(); await new Promise(resolve => server.close(resolve)); rmSync(root, { recursive: true, force: true }); +} diff --git a/src/runtime/hot-sync-manager.ts b/src/runtime/hot-sync-manager.ts index 8441bbe..1bd900a 100644 --- a/src/runtime/hot-sync-manager.ts +++ b/src/runtime/hot-sync-manager.ts @@ -22,6 +22,7 @@ type Metadata = { linked: Record; migrated: Record; mixedVersion: boolean; + recoverySnapshots?: string[]; lastDiscoveryAt?: string; error?: string | null; }; @@ -49,7 +50,11 @@ export class HotSyncManager { private failures = 0; private inflight: Promise | null = null; private stopped = false; + private legacyQueueError: string | null = null; private cursor = 0; + private nextCapabilities = 0; + private nextDiscovery = 0; + private nextLegacyCheck = 0; private active = new Set(); constructor(private readonly config: Config, client?: HostedOnlyBotsClient) { this.namespace = createHmac("sha256", config.nsec).update(`crosshead-profile-v1\n${config.profileName.normalize("NFC")}`).digest("hex"); @@ -82,6 +87,7 @@ export class HotSyncManager { catch { } const id = this.digest(remote ? `git:${remote}` : `local:${raw}`); this.metadata.workspaces[raw] = id; + this.nextLegacyCheck = 0; this.persist(); return id; } @@ -138,22 +144,66 @@ export class HotSyncManager { if (!this.failures) this.nextSync = Math.min(this.nextSync, Date.now() + 1250); } + private captureLegacy(raw: string, runtime: string, checkpoint: Record): void { + const digest = createHash("sha256").update(JSON.stringify(checkpoint)).digest("hex"); + const workspaceId = this.workspace(raw), scope = workspaceHotScope(runtime, workspaceId)!; + this.active.add(scope); + // Identical imports on two devices must have identical signed provenance. + this.replica(scope).capture({ id: `legacy:${digest}`, deviceId: `legacy:${digest}`, runtime, workspaceId, occurredAt: String(checkpoint.occurredAt), goal: String(checkpoint.goal).slice(0, 2000), summary: String(checkpoint.summary).slice(0, 6000), evidence: normalizeTaskEvidence(checkpoint.evidence) ?? undefined }); + this.persist(); + } + /** Import a durable pre-sync queue snapshot before allowing any new upload. + * Stable IDs make an interrupted import safe to repeat before removing its row. */ + setLegacyQueueError(error: string | null): void { this.legacyQueueError = error; } + importQueuedLegacy(state: Record, fallbackRuntime: string, fallbackWorkspace: string | null): void { + const raw = typeof state.workspaceId === "string" ? state.workspaceId : fallbackWorkspace; + if (!raw) throw new Error("Queued hot memory has no workspace; retain it for recovery"); + const runtime = typeof state.runtime === "string" ? state.runtime : fallbackRuntime; + const scope = workspaceHotScope(runtime, this.workspace(raw))!; + const replica = this.replica(scope).inspect(); + // A legacy queue has no reset epoch. Preserve ambiguous snapshots for + // explicit recovery rather than silently discarding or resurrecting them. + if (replica.pendingReset || replica.base.document.epoch > 0) + throw new Error("Queued legacy hot memory needs review after a reset"); + if (!Array.isArray(state.continuityCheckpoints) || !state.continuityCheckpoints.length) + throw new Error("Queued hot memory has no checkpoints; retain it for recovery"); + // An undiscovered remote reset can discard epoch-zero pending captures. + // Retain the exact original locally even after the old queue row is removed. + const digest = createHash("sha256").update(JSON.stringify(state)).digest("hex"); + const recoveryFile = `legacy-recovery-${this.namespace}-${digest}.json`; + if (!(this.metadata.recoverySnapshots ?? []).includes(recoveryFile)) { + if ((this.metadata.recoverySnapshots?.length ?? 0) >= 200) throw new Error("Legacy recovery snapshot limit reached"); + writeHotSyncState(path.join(path.dirname(this.metadataFile), recoveryFile), state); + this.metadata.recoverySnapshots = [...(this.metadata.recoverySnapshots ?? []), recoveryFile]; + this.persist(); + } + for (const value of state.continuityCheckpoints) { + const checkpoint = value as Record; + if (!checkpoint || typeof checkpoint.goal !== "string" || typeof checkpoint.summary !== "string" || typeof checkpoint.occurredAt !== "string" || !Number.isFinite(Date.parse(checkpoint.occurredAt))) + throw new Error("Invalid queued hot-memory checkpoint; retain it for recovery"); + this.captureLegacy(raw, runtime, checkpoint); + } + } reset(scope: string): Record { this.replica(scope).resetLocally(); this.nextSync = 0; return { scope, status: "pending_reset", message: "Cleared on this machine. Other devices clear this scope when they next sync." }; } status(): Record { const scopes = [...this.replicas].map(([scope, r]) => { const x = r.inspect(); return { scope, epoch: x.base.document.epoch, remoteRevision: x.revision, pending: x.pendingCount, pendingReset: x.pendingReset, oldestPendingAt: x.oldestPendingAt, lastSuccessfulSyncAt: x.lastSyncedAt, conflicts: x.local.document.checkpoints.filter(c => c.conflictOf || hotSyncConflictIds(x.local.document).has(c.id)).length, error: x.error ?? (x.local.document.checkpoints.some(c=>workspaceHotScope(c.runtime,c.workspaceId)!==scope) ? "workspace_boundary_invalid" : null), discardedByReset: x.discardedByReset }; }); - return { enabled: true, initialSyncPending: !this.metadata.lastDiscoveryAt, lastSuccessfulDiscoveryAt: this.metadata.lastDiscoveryAt ?? null, pendingCount: scopes.reduce((n, s) => n + s.pending, 0), pendingScopes: scopes.filter(s => s.pending || s.pendingReset).length, state: this.metadata.error || this.metadata.mixedVersion || scopes.some(s=>s.conflicts>0 || s.error) ? "needs_attention" : scopes.some(s => s.pending || s.pendingReset) || !this.metadata.lastDiscoveryAt ? "saved_locally" : "synced", error: this.metadata.error, mixedVersion: this.metadata.mixedVersion, namespace: this.namespace, scopes }; + return { enabled: true, initialSyncPending: !this.metadata.lastDiscoveryAt, lastSuccessfulDiscoveryAt: this.metadata.lastDiscoveryAt ?? null, pendingCount: scopes.reduce((n, s) => n + s.pending, 0), pendingScopes: scopes.filter(s => s.pending || s.pendingReset).length, state: this.legacyQueueError || this.metadata.error || this.metadata.mixedVersion || scopes.some(s=>s.conflicts>0 || s.error || s.discardedByReset>0) ? "needs_attention" : scopes.some(s => s.pending || s.pendingReset) || !this.metadata.lastDiscoveryAt ? "saved_locally" : "synced", error: this.legacyQueueError ?? this.metadata.error, mixedVersion: this.metadata.mixedVersion, legacyRecoverySnapshots: this.metadata.recoverySnapshots ?? [], namespace: this.namespace, scopes }; } tick(): void { if (this.stopped || this.inflight || Date.now() < this.nextSync) return; - this.inflight = this.flush().catch(() => { }).finally(() => { this.inflight = null; }); + this.inflight = this.flush(false).catch(() => { }).finally(() => { this.inflight = null; }); } async stop(): Promise { this.stopped = true; await this.inflight; } - async flush(): Promise { + async flush(forceDiscovery = true): Promise { try { + if (forceDiscovery || Date.now() >= this.nextCapabilities) { const capabilities = await this.client.hotSyncCapabilities(); if (capabilities.version !== HOT_SYNC_PROTOCOL_VERSION || capabilities.enabled !== true) throw new Error("hot_sync_unavailable"); + this.nextCapabilities = Date.now() + 300000; + } + if (forceDiscovery || Date.now() >= this.nextDiscovery) { let cursor: string | null = null; for (let page = 0; page < 2 && !this.stopped; page++) { const listing = await this.client.listHotSync(this.namespace, cursor); @@ -170,6 +220,9 @@ export class HotSyncManager { throw new Error("Invalid scope cursor"); cursor = listing.next_cursor; } + this.nextDiscovery = Date.now() + 30000; + this.metadata.lastDiscoveryAt = new Date().toISOString(); + } const scopes = [...this.replicas.entries()]; // Two slots favor active/dirty scopes, two rotate through discovery/recovery. const preferred = scopes.filter(([scope, r]) => this.active.has(scope) || r.inspect().dirty); @@ -192,7 +245,14 @@ export class HotSyncManager { } this.cursor += 2; let migrated = false; - try { await this.migrateLegacy(); migrated = true; } catch(error) { syncError ??= error; } + if (this.legacyQueueError) syncError ??= new Error(this.legacyQueueError); + else try { + if (forceDiscovery || Date.now() >= this.nextLegacyCheck) { + await this.migrateLegacy(); + this.nextLegacyCheck = Date.now() + 60000; + } + migrated = true; + } catch(error) { syncError ??= error; } // A failed import must not let a first upload create the revision that // would cause the next migration attempt to skip legacy history. if (migrated) for (const replica of selected) { @@ -205,7 +265,6 @@ export class HotSyncManager { if(syncError)throw syncError; this.failures = 0; this.metadata.error = null; - this.metadata.lastDiscoveryAt = new Date().toISOString(); this.persist(); this.nextSync = Date.now() + ([...this.replicas.values()].some(r => r.inspect().dirty) ? 1500 : 15000) + Math.random() * 500; } @@ -213,6 +272,7 @@ export class HotSyncManager { this.failures++; this.metadata.error = error instanceof HostedOnlyBotsRequestError ? error.code ?? `http_${error.status}` : "sync_unavailable"; this.persist(); + if (error instanceof HostedOnlyBotsRequestError && [401, 426, 503].includes(error.status)) this.nextCapabilities = 0; if (error instanceof HostedOnlyBotsRequestError && error.status === 401) this.client.setSession(null); this.nextSync = Date.now() + Math.max(error instanceof HostedOnlyBotsRequestError ? error.retryAfterMs : 0, Math.min(300000, 1000 * 2 ** Math.min(this.failures, 8))) + Math.random() * 1000; @@ -257,7 +317,7 @@ export class HotSyncManager { for (const checkpoint of Array.isArray(state.continuityCheckpoints) ? state.continuityCheckpoints : []) { if (typeof checkpoint.summary !== "string" || typeof checkpoint.goal !== "string" || !Number.isFinite(Date.parse(checkpoint.occurredAt))) continue; - this.capture(scope, { id: `legacy:${createHash("sha256").update(JSON.stringify(checkpoint)).digest("hex")}`, runtime, workspaceId: raw, occurredAt: checkpoint.occurredAt, goal: checkpoint.goal.slice(0, 2000), summary: checkpoint.summary.slice(0, 6000), evidence: normalizeTaskEvidence(checkpoint.evidence) ?? undefined }); + this.captureLegacy(raw, runtime, checkpoint); } } this.metadata.migrated[key] = digest; diff --git a/src/runtime/lifecycle-engine.ts b/src/runtime/lifecycle-engine.ts index 84001c9..a9c517d 100644 --- a/src/runtime/lifecycle-engine.ts +++ b/src/runtime/lifecycle-engine.ts @@ -1262,7 +1262,7 @@ const buildLocalOnlyContextPack = (event: RuntimeLifecycleEvent, packRequest: Re const contextText = hotContextText; const injected = contextText.length > 0 ? 1 : 0; return { - packId: `local-hot-${event.eventId}`, + packId: null, // Local context is not a billable hosted context pack. contextText, fragments: injected > 0 ? [{ id: "local-hot-memory", kind: "hot_memory", content: contextText }] : [], manifest: { @@ -1535,6 +1535,8 @@ export class RuntimeLifecycleManager { private timer: NodeJS.Timeout | null = null; private backgroundOperation: Promise | null = null; private mutation: Promise = Promise.resolve(); + private detachedFlush: Promise | null = null; + private readonly claimedQueueItems = new Set(); constructor(private readonly deps: RuntimeLifecycleDeps) { this.now = deps.now ?? (() => new Date()); @@ -1550,6 +1552,15 @@ export class RuntimeLifecycleManager { return; } this.timer = setInterval(() => { + if (hotSyncEnabled()) { + // Archive transport may be slow; it must not starve hot-memory sync. + void this.withLock(async () => { + if (this.deps.loadContinuityGovernorState().paused) return; + const state = this.loadQueueState(); + this.importQueuedHotMemory(state); + tickHotSyncManagers(this.deps.paths.state); + }).catch(error => this.markFailure(error instanceof Error ? error.message : String(error))); + } if (this.backgroundOperation) return; const operation = this.flushDueItems(false).catch((error) => { this.markFailure(error instanceof Error ? error.message : String(error)); @@ -1572,6 +1583,7 @@ export class RuntimeLifecycleManager { try { await operation; } finally { if (this.backgroundOperation === operation) this.backgroundOperation = null; + await this.detachedFlush; await this.mutation; } } @@ -1620,6 +1632,7 @@ export class RuntimeLifecycleManager { const event = this.hydrateEventFromPriorTurnStart(this.normalizeEvent(request, profileName)); if(hotSyncEnabled() && !localOnlyProfile && event.workspaceId)getHotSyncManager({stateDir:this.deps.paths.state,origin:profile.apiBase,npub:profile.npub,nsec:secret.nsec,profileName}).activate(event.runtime,event.workspaceId); const continuity = this.deps.loadContinuityGovernorState(); + if (hotSyncEnabled() && !continuity.paused) this.importQueuedHotMemory(state); // Apply capture policy before the event log as well as before hot storage. // This also covers a mode/pause change racing a native hook's status read. if (profile.memoryMode === "off" || profile.memoryMode === "manual" || continuity.paused) { @@ -1656,7 +1669,7 @@ export class RuntimeLifecycleManager { this.recordEvent(event); - if ((event.type === "session.started" || event.type === "turn.started") && !localOnlyProfile) { + if ((event.type === "session.started" || event.type === "turn.started") && !localOnlyProfile && !hotSyncEnabled()) { const inboxRefresh = await this.refreshInboxHud(profileName, profile, secret, event); inbox = inboxRefresh.state; if (inboxRefresh.action) { @@ -1725,7 +1738,7 @@ export class RuntimeLifecycleManager { }) ); } else { - if (localOnlyProfile || (hotSyncEnabled() && packRequest.recallPolicy !== "bounded")) { + if (localOnlyProfile || hotSyncEnabled()) { if (!hasHotMemoryOverride(packRequest)) { packRequest.hotMemorySnapshot = localOnlyProfile ? readLocalHotMemory(this.deps.paths.state, HOT_SCOPE_ALL) : getHotSyncManager({stateDir:this.deps.paths.state,origin:profile.apiBase,npub:profile.npub,nsec:secret.nsec,profileName}).read(); } @@ -1847,7 +1860,7 @@ export class RuntimeLifecycleManager { }, }) ); - } else if ((profile.memoryMode === "assist" || profile.memoryMode === "auto") && !localOnlyProfile) { + } else if ((profile.memoryMode === "assist" || profile.memoryMode === "auto") && !localOnlyProfile && !hotSyncEnabled()) { const reflectionRequest = buildReflectionRequest(event); if (reflectionRequest) { const client = this.buildClient(profile, secret); @@ -2214,7 +2227,7 @@ export class RuntimeLifecycleManager { state.processedEventIds = [...state.processedEventIds, event.eventId].slice(-MAX_TRACKED_EVENT_IDS); this.saveQueueState(state); - const processedState = await this.processDueItems(state, false); + const processedState = hotSyncEnabled() ? state : await this.processDueItems(state, false); if (processedState.items.length > this.maxQueueDepth) { warnings.push("Local runtime queue is under backpressure"); @@ -2225,10 +2238,87 @@ export class RuntimeLifecycleManager { } async flushDueItems(force: boolean = true): Promise { - await this.withLock(async () => { + if (!hotSyncEnabled()) { + await this.withLock(async () => { + const state = this.loadQueueState(); + await this.processDueItems(state, force); + }); + return; + } + if (this.detachedFlush) return this.detachedFlush; + const operation = this.flushDetachedQueue(force); + this.detachedFlush = operation; + try { await operation; } + finally { if (this.detachedFlush === operation) this.detachedFlush = null; } + } + + private async flushDetachedQueue(force: boolean): Promise { + const claimed = await this.withLock(async () => { + if (this.deps.loadContinuityGovernorState().paused) return null; const state = this.loadQueueState(); - await this.processDueItems(state, force); + this.importQueuedHotMemory(state); + const now = this.now().getTime(); + const items = state.items.filter(item => { + // Retained hosted snapshots are recovery data, never late legacy uploads. + if (item.kind === "hot_write") { + const profile = this.deps.loadProfile(item.profileName); + if (!profile || !isLocalOnlyRuntimeProfile(profile)) return false; + } + return !this.claimedQueueItems.has(item.id) && (force || Date.parse(item.availableAt) <= now); + }); + for (const item of items) this.claimedQueueItems.add(item.id); + return structuredClone({ items, processedEventIds: [] }) as RuntimeQueueState; }); + if (!claimed) return; + const claimedItems = [...claimed.items]; + const original = new Map(claimedItems.map(item => [item.id, JSON.stringify(item)])); + try { + // These awaits never own the local lifecycle mutation lock. Claimed items + // remain durable, and new captures coalesce only into unclaimed items. + const processed = await this.processDueItems(claimed, true, false); + const remaining = new Map(processed.items.map(item => [item.id, item])); + await this.withLock(async () => { + const current = this.loadQueueState(); + current.items = current.items.flatMap(item => { + const expected = original.get(item.id); + if (!expected || JSON.stringify(item) !== expected) return [item]; + const retry = remaining.get(item.id); + return retry ? [retry] : []; + }); + this.saveQueueState(current); + }); + } finally { + await this.withLock(async () => { for (const item of claimedItems) this.claimedQueueItems.delete(item.id); }); + } + } + + private importQueuedHotMemory(state: RuntimeQueueState): void { + const imported = new Set(); + const errors = new Map, string | null>(); + for (const item of state.items) { + if (item.kind !== "hot_write" || this.claimedQueueItems.has(item.id)) continue; + try { + const { profile, secret } = this.resolveProfile(item.profileName); + if (isLocalOnlyRuntimeProfile(profile)) continue; + const manager = getHotSyncManager({ stateDir: this.deps.paths.state, origin: profile.apiBase, npub: profile.npub, nsec: secret.nsec, profileName: item.profileName }); + if (!errors.has(manager)) errors.set(manager, null); + try { + manager.importQueuedLegacy(item.payload.hotState, item.payload.scope ?? "shared", item.workspaceId); + imported.add(item.id); + } catch (error) { + errors.set(manager, error instanceof Error ? error.message : "Legacy queue needs recovery"); + } + } catch (error) { + // A removed/broken profile must not prevent unrelated local hooks. + this.deps.writeLog("hot_sync_legacy_queue_retained", { queueItemId: item.id, profileName: item.profileName }); + } + } + for (const [manager, error] of errors) manager.setLegacyQueueError(error); + if (imported.size) { + // Each imported checkpoint is fsynced before its old queue entry is removed. + state.items = state.items.filter(item => !imported.has(item.id)); + this.saveQueueState(state); + } } private async withLock(fn: () => Promise): Promise { @@ -2824,7 +2914,7 @@ export class RuntimeLifecycleManager { const now = this.now(); const dueAt = options.immediate ? now.toISOString() : new Date(now.getTime() + this.hotWriteDebounceMs).toISOString(); const existing = state.items.find( - (item): item is Extract => item.kind === "hot_write" && item.key === key + (item): item is Extract => item.kind === "hot_write" && item.key === key && !this.claimedQueueItems.has(item.id) ); if (existing) { existing.eventId = event.eventId; @@ -2864,7 +2954,7 @@ export class RuntimeLifecycleManager { const now = this.now(); const dueAt = new Date(now.getTime() + this.coldBatchWindowMs).toISOString(); const existing = state.items.find( - (item): item is Extract => item.kind === "cold_batch" && item.key === key + (item): item is Extract => item.kind === "cold_batch" && item.key === key && !this.claimedQueueItems.has(item.id) ); if (existing) { existing.eventId = event.eventId; @@ -2917,10 +3007,10 @@ export class RuntimeLifecycleManager { return item; } - private async processDueItems(state: RuntimeQueueState, force: boolean): Promise { + private async processDueItems(state: RuntimeQueueState, force: boolean, persistState = true): Promise { const continuity = this.deps.loadContinuityGovernorState(); if (continuity.paused) { - this.saveQueueState(state); + if (persistState) this.saveQueueState(state); return state; } @@ -2931,6 +3021,7 @@ export class RuntimeLifecycleManager { const pending = state.items.filter((item) => !ready.includes(item)); for (const item of ready) { + if (this.deps.loadContinuityGovernorState().paused) { remaining.push(item); continue; } try { const result = await this.processQueueItem(item); const warmDecision = @@ -3065,7 +3156,7 @@ export class RuntimeLifecycleManager { } state.items = [...pending, ...remaining]; - this.saveQueueState(state); + if (persistState) this.saveQueueState(state); return state; } diff --git a/tests/runtime/hot-sync-acceptance.test.ts b/tests/runtime/hot-sync-acceptance.test.ts index 7729a67..b24409e 100644 --- a/tests/runtime/hot-sync-acceptance.test.ts +++ b/tests/runtime/hot-sync-acceptance.test.ts @@ -54,6 +54,70 @@ const entries = (m: HotSyncManager, inspect = false) => (m.read("all", inspect). content: string; }>).flatMap(e => JSON.parse(e.content).continuityCheckpoints as HotSyncCheckpoint[]); describe("local-first hot-memory acceptance", () => { + it("bounds background capability, scope-discovery and legacy-read traffic during repeated captures", async () => { + const cloud = new Cloud(), id = identity(), client = cloud.client(id.npub); + const capabilities = vi.spyOn(client, "hotSyncCapabilities"), discovery = vi.spyOn(client, "listHotSync"), legacy = vi.spyOn(client, "readHotMemory"); + const m = new HotSyncManager({ origin: "https://crosshead.sh", ...id, profileName: "default", stateDir: home() }, client); + for (const name of ["first", "second", "third"]) { m.capture("codex", checkpoint(name)); await m.flush(false); } + expect(capabilities).toHaveBeenCalledTimes(1); + expect(discovery).toHaveBeenCalledTimes(1); + expect(legacy).toHaveBeenCalledTimes(1); + expect(entries(m)).toHaveLength(3); + expect(m.status().pendingCount).toBe(0); + m.workspace("repo:/new-project"); await m.flush(false); + expect(legacy).toHaveBeenCalledTimes(2); // New workspaces must still migrate before upload. + await m.flush(); // Explicit operator refresh bypasses discovery caches. + expect(capabilities).toHaveBeenCalledTimes(2); + expect(discovery).toHaveBeenCalledTimes(2); + expect(legacy).toHaveBeenCalledTimes(3); + }); + + it("imports identical queued legacy findings on two devices without false conflicts", async () => { + const cloud = new Cloud(), id = identity(); + const config = { origin: "https://crosshead.sh", ...id, profileName: "default" }; + const a = new HotSyncManager({ ...config, stateDir: home() }, cloud.client(id.npub)); + const b = new HotSyncManager({ ...config, stateDir: home() }, cloud.client(id.npub)); + const legacy = { runtime: "codex", workspaceId: "repo:/project", continuityCheckpoints: [{ goal: "Recover", summary: "One shared legacy finding", occurredAt: "2026-11-03T00:00:00Z" }] }; + a.importQueuedLegacy(legacy, "codex", null); + b.importQueuedLegacy(legacy, "codex", null); + await a.flush(); await b.flush(); await a.flush(); + expect(entries(a)).toHaveLength(1); + expect(entries(b)).toHaveLength(1); + expect(entries(a)[0]?.summary).toBe("One shared legacy finding"); + expect(entries(a, true).some(c => c.conflictOf)).toBe(false); + }); + it("keeps a durable recovery copy when first discovery reveals a remote reset", async () => { + const cloud = new Cloud(), id = identity(); + const config = { origin: "https://crosshead.sh", ...id, profileName: "default" }; + const a = new HotSyncManager({ ...config, stateDir: home() }, cloud.client(id.npub)); + a.capture("codex", checkpoint("old")); + const scope = (a.status().scopes as Array<{scope:string}>)[0]!.scope; + a.reset(scope); await a.flush(); + const stateDir = home(), b = new HotSyncManager({ ...config, stateDir }, cloud.client(id.npub)); + const legacy = { runtime: "codex", workspaceId: "repo:/project", continuityCheckpoints: [{ goal: "Recover", summary: "Ambiguous work survives reset discovery", occurredAt: "2026-11-03T00:00:00Z" }] }; + b.importQueuedLegacy(legacy, "codex", null); + await b.flush(); + expect(entries(b)).toHaveLength(0); // Never resurrect stale memory automatically. + const files = b.status().legacyRecoverySnapshots as string[]; + expect(files).toHaveLength(1); + expect(JSON.parse(readFileSync(path.join(hotSyncTenantDirectory(stateDir, config.origin, id.npub), files[0]!), "utf8"))).toEqual(legacy); + expect(b.status().state).toBe("needs_attention"); + }); + + it("preserves unsupported or reset-ambiguous queued snapshots for explicit recovery", async () => { + const cloud = new Cloud(), id = identity(); + const m = new HotSyncManager({ origin: "https://crosshead.sh", ...id, profileName: "default", stateDir: home() }, cloud.client(id.npub)); + expect(() => m.importQueuedLegacy({ currentTask: "Unsent task" }, "codex", "repo:/project")).toThrow("no checkpoints"); + const legacy = { runtime: "codex", workspaceId: "repo:/project", continuityCheckpoints: [{ goal: "Recover", summary: "Retain after reset", occurredAt: "2026-11-03T00:00:00Z" }] }; + m.importQueuedLegacy(legacy, "codex", null); + const scope = (m.status().scopes as Array<{scope: string}>)[0]!.scope; + m.reset(scope); + expect(() => m.importQueuedLegacy(legacy, "codex", null)).toThrow("needs review after a reset"); + await m.flush(); + expect(() => m.importQueuedLegacy(legacy, "codex", null)).toThrow("needs review after a reset"); + expect(entries(m)).toHaveLength(0); + }); + it("maps sanitized remotes and explicit links across paths, while isolating profiles and identities", async () => { expect(canonicalGitRemote("git@github.com:FriendMachine/crosshead.git")).toBe("github.com/FriendMachine/crosshead"); expect(canonicalGitRemote("https://user:secret@github.com/FriendMachine/crosshead.git?token=secret")).toBe("github.com/FriendMachine/crosshead"); diff --git a/tests/runtime/hot-sync-hook-path.test.ts b/tests/runtime/hot-sync-hook-path.test.ts new file mode 100644 index 0000000..9f51086 --- /dev/null +++ b/tests/runtime/hot-sync-hook-path.test.ts @@ -0,0 +1,169 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { randomUUID } from "node:crypto"; +import { generateSecretKey, getPublicKey, nip19 } from "nostr-tools"; +import { HostRuntimeAdapterBase, type HostAdapterIdentity } from "../../src/runtime/host-adapter-base"; +import { RuntimeLifecycleManager } from "../../src/runtime/lifecycle-engine"; +import { HostedOnlyBotsClient } from "../../src/runtime/hosted-client"; +import { getHotSyncManager, stopHotSyncManagers } from "../../src/runtime/hot-sync-manager"; +import { buildContinuityGovernorState } from "../../src/runtime/continuity-governor"; +import { buildDefaultContinuityPolicy } from "../../src/runtime/continuity-policy"; +import type { LocalRuntimeClient } from "../../src/runtime/local-client"; + +const homes: string[] = []; +const releases: Array<() => void> = []; +const managers: RuntimeLifecycleManager[] = []; +class Adapter extends HostRuntimeAdapterBase { + protected readonly identity: HostAdapterIdentity; + protected readonly config = { profileName: "default", hostHome: "/fixture-host", adapterVersion: "fixture" }; + constructor(protected readonly client: LocalRuntimeClient, runtime: string) { super(); this.identity = { runtimeId: runtime, wrapperVersionFallback: "fixture", requestLabel: runtime }; } + protected async attachWorkspace(): Promise {} +} +const deadline = async (work: Promise): Promise => { + let timer: ReturnType; + try { return await Promise.race([work, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error("hook waited for the hosted service")), 500); })]); } + finally { clearTimeout(timer!); } +}; +const harness = (runtime = "codex", coldSync = false) => { + vi.stubEnv("CROSSHEAD_HOT_SYNC_ENABLED", "true"); + let blocked = true; + const pending = new Set<(reason: Error) => void>(); + const fetcher = vi.fn(() => blocked ? new Promise((_, reject) => pending.add(reject)) : Promise.reject(new Error("fixture cloud offline"))); + vi.stubGlobal("fetch", fetcher); + const release = () => { blocked = false; for (const reject of pending) reject(new Error("fixture cloud offline")); pending.clear(); }; + releases.push(release); + const stateDir = mkdtempSync(path.join(os.tmpdir(), "xhd-hook-")); homes.push(stateDir); + const key = generateSecretKey(), npub = nip19.npubEncode(getPublicKey(key)), nsec = nip19.nsecEncode(key); + const profile = { name: "default", apiBase: "https://crosshead.sh", npub, memoryMode: coldSync ? "auto" as const : "assist" as const, features: { coldSync, autoWarmPromotion: false } }; + const lifecycle = new RuntimeLifecycleManager({ workerIntervalMs: 20, paths: { state: stateDir }, loadDefaultProfileName: () => "default", loadProfile: name => name === "default" ? profile : null, + loadSecret: () => ({ nsec, session: null }), saveSecret: () => {}, loadHealth: () => ({ queueDepth: 0, lastSuccessfulSyncAt: null, lastError: null }), saveHealth: () => {}, + loadContinuityGovernorState: () => buildContinuityGovernorState(buildDefaultContinuityPolicy(), []), writeLog: () => {} }); + managers.push(lifecycle); + const sync = getHotSyncManager({ stateDir, origin: profile.apiBase, npub, nsec, profileName: "default" }); + const client = { readHotMemory: async () => sync.read(), emitEvent: (event: Parameters[0]) => lifecycle.handleEvent({ ...event, eventId: event.eventId ?? randomUUID() }) } as unknown as LocalRuntimeClient; + return { stateDir, lifecycle, sync, adapter: new Adapter(client, runtime), fetcher, release }; +}; +afterEach(async () => { + for (const release of releases.splice(0)) release(); + for (const manager of managers.splice(0)) await manager.stop(); + for (const home of homes.splice(0)) { await stopHotSyncManagers(home); rmSync(home, { recursive: true, force: true }); } + vi.unstubAllEnvs(); vi.unstubAllGlobals(); vi.restoreAllMocks(); +}); + +describe("real adapter local hot-memory path", () => { + it.each(["codex", "claude-code", "pi", "hermes", "opencode"])("prepares and durably completes %s while hosted requests hang", async runtime => { + const h = harness(runtime); + const input = { workspaceRoot: "/fixture-project", sessionId: "session", turnId: "first", prompt: "Repair the queue" }; + const prepared = await deadline(h.adapter.prepareTurn(input)); + expect(prepared.packId).toBeNull(); + await deadline(h.adapter.completeTurn({ ...input, packId: prepared.packId, goal: input.prompt, summary: "Fsync the queue before acknowledging capture." })); + expect(h.sync.status().pendingCount).toBe(1); + const continued = await deadline(h.adapter.prepareTurn({ ...input, turnId: "second", prompt: "Continue queue repair" })); + expect(continued.contextText).toContain("Fsync the queue before acknowledging capture."); + expect(h.fetcher).not.toHaveBeenCalled(); + }); + it("does not hold the lifecycle lock during a previously due archive upload", async () => { + const h = harness(); + const now = new Date().toISOString(); + const item = { id: "old-archive", key: "cold:default:session", kind: "cold_batch", profileName: "default", eventId: "old", sessionId: "session", workspaceId: "repo:/fixture-project", createdAt: now, availableAt: now, attempts: 0, payload: { items: [{ kind: "note", content: "old archive fixture" }] } }; + writeFileSync(path.join(h.stateDir, "queue.json"), JSON.stringify({ items: [item], processedEventIds: [] })); + const flushing = h.lifecycle.flushDueItems(); + try { + await vi.waitFor(() => expect(h.fetcher).toHaveBeenCalled(), { timeout: 1000 }); + const input = { workspaceRoot: "/fixture-project", sessionId: "session", turnId: "new", prompt: "Continue" }; + const prepared = await deadline(h.adapter.prepareTurn(input)); + await deadline(h.adapter.completeTurn({ ...input, packId: prepared.packId, goal: input.prompt, summary: "The new finding survives a blocked archive." })); + expect(h.sync.status().pendingCount).toBe(1); + // The durable processed-event IDs written by new hooks must survive the + // older worker's acknowledgment/retry commit. + const before = JSON.parse(readFileSync(path.join(h.stateDir, "queue.json"), "utf8")); + h.release(); await flushing; + const after = JSON.parse(readFileSync(path.join(h.stateDir, "queue.json"), "utf8")); + expect(after.processedEventIds).toEqual(before.processedEventIds); + expect(after.items).toEqual(expect.arrayContaining([expect.objectContaining({ id: "old-archive", attempts: 1 })])); + } finally { h.release(); await flushing; } + }); + it("imports unsent upgrade snapshots locally before new captures, and repeats safely after a crash", async () => { + const h = harness(); + const checkpoint = { goal: "Recover upgrade", summary: "The unsent pre-upgrade finding", occurredAt: new Date().toISOString() }; + const item = { id: "upgrade", kind: "hot_write", profileName: "default", workspaceId: "repo:/fixture-project", payload: { scope: "codex", hotState: { runtime: "codex", workspaceId: "repo:/fixture-project", continuityCheckpoints: [checkpoint] } } }; + const queue = path.join(h.stateDir, "queue.json"); + const original = JSON.stringify({ items: [item], processedEventIds: [] }); + writeFileSync(queue, original); + const input = { workspaceRoot: "/fixture-project", sessionId: "session", turnId: "upgrade", prompt: "Continue" }; + await deadline(h.adapter.prepareTurn(input)); + expect(h.sync.status().pendingCount).toBe(1); + // Simulate a crash after the outbox fsync but before legacy-row removal. + writeFileSync(queue, original); + await deadline(h.adapter.completeTurn({ ...input, packId: null, goal: input.prompt, summary: "A new post-upgrade finding" })); + expect(h.sync.status().pendingCount).toBe(2); + const resumed = await deadline(h.adapter.prepareTurn({ ...input, turnId: "upgraded" })); + expect(JSON.stringify(h.sync.read())).toContain(checkpoint.summary); + expect(resumed.contextText).toContain("A new post-upgrade finding"); + expect(JSON.parse(readFileSync(queue, "utf8")).items).toEqual([]); + expect(h.fetcher).not.toHaveBeenCalled(); + }); + it("acknowledges an uploaded archive without removing the same-key arrivals or processed events", async () => { + const h = harness("codex", true); + let release!: () => void; + const upload = vi.spyOn(HostedOnlyBotsClient.prototype, "ingestColdMemory").mockImplementationOnce(() => new Promise(resolve => { release = () => resolve({}); })); + const input = { workspaceRoot: "/fixture-project", sessionId: "session", turnId: "one", prompt: "Continue" }; + await h.adapter.completeTurn({ ...input, packId: null, goal: input.prompt, summary: "First archived finding", transcript: [{ role: "assistant", content: "First archived finding" }] }); + const file = path.join(h.stateDir, "queue.json"); + const first = JSON.parse(readFileSync(file, "utf8")); + expect(first.items).toHaveLength(1); + const flushing = h.lifecycle.flushDueItems(); + try { + await vi.waitFor(() => expect(upload).toHaveBeenCalledOnce()); + await deadline(h.adapter.completeTurn({ ...input, turnId: "two", packId: null, goal: input.prompt, summary: "Second archived finding", transcript: [{ role: "assistant", content: "Second archived finding" }] })); + const before = JSON.parse(readFileSync(file, "utf8")); + expect(before.items).toHaveLength(2); + expect(before.items[0].key).toBe(before.items[1].key); + release(); await flushing; + const after = JSON.parse(readFileSync(file, "utf8")); + expect(after.items.map((i: {id: string}) => i.id)).toEqual([before.items[1].id]); + expect(after.processedEventIds).toEqual(before.processedEventIds); + } finally { release?.(); h.release(); await flushing; } + }); + + it("retains an unimportable row without blocking healthy hooks or uploading it as legacy data", async () => { + const h = harness(); + const file = path.join(h.stateDir, "queue.json"); + const rows = [ + { id: "unsupported", kind: "hot_write", profileName: "default", workspaceId: null, payload: { hotState: { currentTask: "Keep this task" } } }, + { id: "removed-profile", kind: "hot_write", profileName: "removed", payload: { hotState: {} } }, + ]; + writeFileSync(file, JSON.stringify({ items: rows, processedEventIds: [] })); + await deadline(h.adapter.completeTurn({ workspaceRoot: "/fixture-project", sessionId: "session", turnId: "healthy", packId: null, goal: "Continue", summary: "Healthy local capture" })); + expect(h.sync.status().pendingCount).toBe(1); + expect(h.sync.status().state).toBe("needs_attention"); + await deadline(h.lifecycle.flushDueItems()); + expect(JSON.parse(readFileSync(file, "utf8")).items).toEqual(rows); + h.release(); + }); + it("continues hot sync while an archive upload remains blocked", async () => { + const h = harness(); + vi.spyOn(HostedOnlyBotsClient.prototype, "hotSyncCapabilities").mockResolvedValue({ enabled: true, version: 2 }); + vi.spyOn(HostedOnlyBotsClient.prototype, "listHotSync").mockResolvedValue({ scopes: [], next_cursor: null }); + vi.spyOn(HostedOnlyBotsClient.prototype, "readHotMemory").mockResolvedValue({ scopes: [] }); + vi.spyOn(HostedOnlyBotsClient.prototype, "readHotSync").mockImplementation(async (namespace, scope) => ({ namespace, scope, epoch: 0, revision: 0, content: null })); + const write = vi.spyOn(HostedOnlyBotsClient.prototype, "writeHotSync").mockImplementation(async (base, content) => ({ ...base, revision: base.revision + 1, content })); + let release!: () => void; + const archive = vi.spyOn(HostedOnlyBotsClient.prototype, "ingestColdMemory").mockImplementation(() => new Promise(resolve => { release = () => resolve({}); })); + const now = new Date().toISOString(); + writeFileSync(path.join(h.stateDir, "queue.json"), JSON.stringify({ items: [{ id: "archive", kind: "cold_batch", profileName: "default", eventId: "archive", availableAt: now, attempts: 0, payload: { items: [{ role: "assistant", content: "Archive", timestamp: now }] } }], processedEventIds: [] })); + h.lifecycle.start(); + try { + await vi.waitFor(() => expect(archive).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(h.sync.status().initialSyncPending).toBe(false)); + await deadline(h.adapter.completeTurn({ workspaceRoot: "/fixture-project", sessionId: "session", turnId: "during-archive", packId: null, goal: "Continue", summary: "Sync before archive finishes" })); + expect(h.sync.status().pendingCount).toBe(1); + await vi.waitFor(() => expect(h.sync.status().pendingCount).toBe(0), { timeout: 4000 }); + expect(write).toHaveBeenCalled(); + expect(JSON.parse(readFileSync(path.join(h.stateDir, "queue.json"), "utf8")).items).toHaveLength(1); + } finally { release?.(); h.release(); await h.lifecycle.stop(); } + }); + +}); From 9affb69244136c91e4c21cb786c0ae5ff94618a1 Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 15:07:43 -0500 Subject: [PATCH 2/8] fix: preserve migration ordering during workspace discovery --- src/runtime/hot-sync-manager.ts | 7 ++++++- tests/runtime/hot-sync-acceptance.test.ts | 21 +++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/src/runtime/hot-sync-manager.ts b/src/runtime/hot-sync-manager.ts index 1bd900a..1b3e322 100644 --- a/src/runtime/hot-sync-manager.ts +++ b/src/runtime/hot-sync-manager.ts @@ -55,6 +55,7 @@ export class HotSyncManager { private nextCapabilities = 0; private nextDiscovery = 0; private nextLegacyCheck = 0; + private workspaceGeneration = 0; private active = new Set(); constructor(private readonly config: Config, client?: HostedOnlyBotsClient) { this.namespace = createHmac("sha256", config.nsec).update(`crosshead-profile-v1\n${config.profileName.normalize("NFC")}`).digest("hex"); @@ -88,6 +89,7 @@ export class HotSyncManager { const id = this.digest(remote ? `git:${remote}` : `local:${raw}`); this.metadata.workspaces[raw] = id; this.nextLegacyCheck = 0; + this.workspaceGeneration++; this.persist(); return id; } @@ -100,6 +102,8 @@ export class HotSyncManager { throw new Error("Workspace already contains hot memory; reset its scopes before relinking"); this.metadata.workspaces[raw] = id; this.metadata.linked[raw] = label; + this.nextLegacyCheck = 0; + this.workspaceGeneration++; this.persist(); return { workspaceId: id, workspaceRoot: path.resolve(root) }; } @@ -248,8 +252,9 @@ export class HotSyncManager { if (this.legacyQueueError) syncError ??= new Error(this.legacyQueueError); else try { if (forceDiscovery || Date.now() >= this.nextLegacyCheck) { + const generation = this.workspaceGeneration; await this.migrateLegacy(); - this.nextLegacyCheck = Date.now() + 60000; + this.nextLegacyCheck = generation === this.workspaceGeneration ? Date.now() + 60000 : 0; } migrated = true; } catch(error) { syncError ??= error; } diff --git a/tests/runtime/hot-sync-acceptance.test.ts b/tests/runtime/hot-sync-acceptance.test.ts index b24409e..5af39f7 100644 --- a/tests/runtime/hot-sync-acceptance.test.ts +++ b/tests/runtime/hot-sync-acceptance.test.ts @@ -72,6 +72,27 @@ describe("local-first hot-memory acceptance", () => { expect(legacy).toHaveBeenCalledTimes(3); }); + it("rechecks migration if a new workspace appears during a background legacy read", async () => { + const cloud = new Cloud(), id = identity(), client = cloud.client(id.npub); + const state = (workspaceId: string, summary: string) => ({ runtime: "codex", workspaceId, continuityCheckpoints: [{ goal: "Recover", summary, occurredAt: "2026-11-03T00:00:00Z" }] }); + cloud.legacy = { scopes: [ + { scope: "new", integrity: "verified", content: JSON.stringify(state("repo:/new", "Legacy new workspace")) }, + { scope: "known", integrity: "verified", content: JSON.stringify(state("repo:/known", "Legacy known workspace")) }, + ] }; + const m = new HotSyncManager({ origin: "https://crosshead.sh", ...id, profileName: "default", stateDir: home() }, client); + m.workspace("repo:/known"); + const original = client.readHotSync.bind(client); let release!: () => void; + const read = vi.spyOn(client, "readHotSync").mockImplementationOnce((...args) => new Promise(resolve => { release = () => resolve(original(...args)); })); + const flushing = m.flush(false); + try { + await vi.waitFor(() => expect(read).toHaveBeenCalled()); + m.capture("codex", { ...checkpoint("new-capture"), workspaceId: "repo:/new" }); + release(); await flushing; await m.flush(false); + expect(entries(m).some(c => c.summary === "Legacy new workspace")).toBe(true); + expect(entries(m).some(c => c.id === "new-capture")).toBe(true); + } finally { release?.(); await flushing; } + }); + it("imports identical queued legacy findings on two devices without false conflicts", async () => { const cloud = new Cloud(), id = identity(); const config = { origin: "https://crosshead.sh", ...id, profileName: "default" }; From afa7b7c07f789558905c41b14b28614dcd62f629 Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 15:10:57 -0500 Subject: [PATCH 3/8] fix: bind legacy migration to the checked workspace scope --- src/runtime/hot-sync-manager.ts | 3 +++ tests/runtime/hot-sync-acceptance.test.ts | 19 +++++++++++++++++++ 2 files changed, 22 insertions(+) diff --git a/src/runtime/hot-sync-manager.ts b/src/runtime/hot-sync-manager.ts index 1b3e322..bffdce8 100644 --- a/src/runtime/hot-sync-manager.ts +++ b/src/runtime/hot-sync-manager.ts @@ -317,6 +317,9 @@ export class HotSyncManager { // Never let a legacy import repopulate or block an explicitly reset scope. if(this.replica(scope).inspect().pendingReset)continue; const remote = await this.client.readHotSync(this.namespace, scope); + // Linking can change the mapping while the cloud read is pending. Its + // absence/reset proof is valid only for the scope that was read. + if (workspaceHotScope(runtime, this.workspace(raw)) !== scope) continue; if(this.replica(scope).inspect().pendingReset)continue; if (remote.revision === 0 && remote.epoch === 0) { for (const checkpoint of Array.isArray(state.continuityCheckpoints) ? state.continuityCheckpoints : []) { diff --git a/tests/runtime/hot-sync-acceptance.test.ts b/tests/runtime/hot-sync-acceptance.test.ts index 5af39f7..267d651 100644 --- a/tests/runtime/hot-sync-acceptance.test.ts +++ b/tests/runtime/hot-sync-acceptance.test.ts @@ -93,6 +93,25 @@ describe("local-first hot-memory acceptance", () => { } finally { release?.(); await flushing; } }); + it("does not use an old-scope absence proof after a workspace is relinked", async () => { + const cloud = new Cloud(), id = identity(), config = { origin: "https://crosshead.sh", ...id, profileName: "default" }; + const other = new HotSyncManager({ ...config, stateDir: home() }, cloud.client(id.npub)); + other.link("/remote-copy", "shared-target"); other.capture("codex", { ...checkpoint("old"), workspaceId: "repo:/remote-copy" }); + const target = (other.status().scopes as Array<{scope:string}>)[0]!.scope; + other.reset(target); await other.flush(); + cloud.legacy = { scopes: [{ scope: "codex", integrity: "verified", content: JSON.stringify({ runtime: "codex", workspaceId: "repo:/project", continuityCheckpoints: [{ goal: "Recover", summary: "Must not revive after target reset", occurredAt: "2026-11-03T00:00:00Z" }] }) }] }; + const client = cloud.client(id.npub), m = new HotSyncManager({ ...config, stateDir: home() }, client); + m.workspace("repo:/project"); const original = client.readHotSync.bind(client); let release!: () => void; + const read = vi.spyOn(client, "readHotSync").mockImplementationOnce((...args) => new Promise(resolve => { release = () => resolve(original(...args)); })); + const flushing = m.flush(false); + try { + await vi.waitFor(() => expect(read).toHaveBeenCalled()); + m.link("/project", "shared-target"); release(); await flushing; await m.flush(false); + expect(entries(m)).toHaveLength(0); + expect((await original(m.namespace, target)).epoch).toBe(1); + } finally { release?.(); await flushing; } + }); + it("imports identical queued legacy findings on two devices without false conflicts", async () => { const cloud = new Cloud(), id = identity(); const config = { origin: "https://crosshead.sh", ...id, profileName: "default" }; From 27cb3432feffe6eca6be52498acd4396276bfea9 Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 15:31:20 -0500 Subject: [PATCH 4/8] ci: run the pinned coding hosts native installers --- .github/workflows/coding-hosts.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/coding-hosts.yml b/.github/workflows/coding-hosts.yml index 2c50e91..1fc8d00 100644 --- a/.github/workflows/coding-hosts.yml +++ b/.github/workflows/coding-hosts.yml @@ -34,8 +34,8 @@ jobs: cache: npm - run: npm ci - run: npm run release:build - - name: Install pinned real coding hosts - run: npm install --prefix "$RUNNER_TEMP/coding-hosts" --ignore-scripts --no-audit --no-fund @openai/codex@0.164.0 @anthropic-ai/claude-code@2.1.319 + - name: Install pinned real coding hosts with required native installers + run: npm install --prefix "$RUNNER_TEMP/coding-hosts" --no-audit --no-fund @openai/codex@0.164.0 @anthropic-ai/claude-code@2.1.319 - name: Exercise clean candidate install and real host hooks env: CROSSHEAD_TEST_CLAUDE_BIN: ${{ runner.temp }}/coding-hosts/node_modules/.bin/claude From 06c6f0ce06ece315cdb686b567994e535e178b64 Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 15:57:45 -0500 Subject: [PATCH 5/8] ci: canonicalize fixture homes on macOS and preserve failures --- scripts/smoke-coding-hosts.mjs | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/scripts/smoke-coding-hosts.mjs b/scripts/smoke-coding-hosts.mjs index 1a257c2..c4abd90 100644 --- a/scripts/smoke-coding-hosts.mjs +++ b/scripts/smoke-coding-hosts.mjs @@ -6,12 +6,12 @@ import { spawn } from 'node:child_process'; import { createServer } from 'node:http'; import { createRequire } from 'node:module'; import { createHash } from 'node:crypto'; -import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, rmSync } from 'node:fs'; +import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, rmSync, realpathSync } from 'node:fs'; import os from 'node:os'; import path from 'node:path'; const require = createRequire(import.meta.url); const tar = require('tar'); -const root = mkdtempSync(path.join(os.tmpdir(), 'crosshead-coding-hosts-')); +const root = realpathSync(mkdtempSync(path.join(os.tmpdir(), 'crosshead-coding-hosts-'))); const output = path.resolve(process.env.CROSSHEAD_TEST_LOG_DIR ?? '.coding-host-test-logs'); mkdirSync(output, { recursive: true }); const installed = path.join(root, 'installed'), home = path.join(root, 'runtime'), project = path.join(root, 'project'); @@ -74,6 +74,7 @@ const run = async (name, binary, args, cwd = project) => { assert.equal(code, 0, `${name} exited ${code}; inspect ${output}/${name}.log`); return stdout; }; const results = { platform: process.platform, checks: [], fixture: true }; +let primaryFailure = null; try { const releaseDir = path.resolve(process.argv[2] ?? 'release'); const manifest = JSON.parse(readFileSync(path.join(releaseDir, 'crosshead-release.json'), 'utf8')); @@ -125,7 +126,15 @@ try { results.releaseId = manifest.releaseId; results.source = manifest.source; writeFileSync(path.join(output, 'result.json'), JSON.stringify(results, null, 2) + '\n'); console.log(JSON.stringify(results)); -} finally { +} catch (error) { primaryFailure = error; throw error; } finally { writeFileSync(path.join(output, 'provider-observations.json'), JSON.stringify(requests.map(({stage, path: route, body}) => ({ stage, route, hasFinding: JSON.stringify(body).includes(marker), hasCaptureContract: JSON.stringify(body).includes('Crosshead hooks automatically capture') })), null, 2) + '\n'); - await cleanup(); await new Promise(resolve => server.close(resolve)); rmSync(root, { recursive: true, force: true }); + let cleaned = false; + try { await cleanup(); cleaned = true; } + catch (error) { + writeFileSync(path.join(output, 'cleanup-error.log'), `${error.message}\nRetained fixture home: ${root}\n`); + if (!primaryFailure) throw error; + } finally { + await new Promise(resolve => server.close(resolve)); + if (cleaned) rmSync(root, { recursive: true, force: true }); + } } From 675261313f8c405fff98e54b4a5830a53efee2d5 Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 16:14:46 -0500 Subject: [PATCH 6/8] feat: default to local memory and sign explicit erasure requests --- docs/HOT-SYNC.md | 28 +++++++++++++++---- scripts/profile-hot-sync.mjs | 10 +++++-- scripts/smoke-coding-hosts.mjs | 2 +- src/runtime/cli.ts | 6 +++- src/runtime/cli/organization.ts | 9 ++++++ src/runtime/erasure-proof.ts | 23 +++++++++++++++ src/runtime/hot-sync-manager.ts | 2 +- src/runtime/local-runtime.ts | 12 ++++++++ tests/runtime/erasure-proof.test.ts | 12 ++++++++ .../runtime/hosted-lifecycle-security.test.ts | 3 ++ tests/runtime/hot-sync-hook-path.test.ts | 7 ++++- 11 files changed, 102 insertions(+), 12 deletions(-) create mode 100644 src/runtime/erasure-proof.ts create mode 100644 tests/runtime/erasure-proof.test.ts diff --git a/docs/HOT-SYNC.md b/docs/HOT-SYNC.md index 6834642..837bec2 100644 --- a/docs/HOT-SYNC.md +++ b/docs/HOT-SYNC.md @@ -1,10 +1,10 @@ # Local hot memory with background sync -Local hot sync is available in the v0.5.0 alpha series. Enable -`CROSSHEAD_HOT_SYNC_ENABLED=true` in the daemon's environment, with a server -running migration 0034 and the same flag. All participating devices must use -protocol 2. Production capacity validation is required before changing the -launch default or advertising multi-device recovery as generally available. +Local hot sync is the default beginning with v0.5.0-alpha.7. The server must +run migration 0034 and enable `CROSSHEAD_HOT_SYNC_ENABLED=true`; participating +devices use protocol 2. The daemon owns this decision. Set +`CROSSHEAD_HOT_SYNC_ENABLED=false` in its environment only for an intentional +rollback to the previous hosted hook path. ## What happens locally @@ -152,3 +152,21 @@ allowance before authentication (including invalid credentials) and a verified identity allowance after authentication; creating sessions cannot multiply it. These counters are per server process, as with the existing limiter; aggregate multi-instance capacity still requires deployment measurement. + +## Requesting hosted private-memory deletion + +Email `support@crosshead.sh` with the agent's public `npub`. Support sends a +short-lived deletion challenge. Run the supplied command on the machine holding +that identity: + +```sh +crosshead sign-deletion-challenge '' --confirm erase-private-memory --json +``` + +The CLI checks the challenge's purpose, identity, server and expiry against the +local profile. Send the resulting signed proof to support; the private key stays +on your machine. This authorizes deletion of that identity's hosted private +memory across profiles and namespaces, and permanently disables the cloud +identity so an old device cannot upload it again. Local files remain on your +machines. Shared organizations, public publications and required financial +records have separate ownership and retention rules. diff --git a/scripts/profile-hot-sync.mjs b/scripts/profile-hot-sync.mjs index 62cc042..a505e92 100644 --- a/scripts/profile-hot-sync.mjs +++ b/scripts/profile-hot-sync.mjs @@ -28,11 +28,12 @@ globalThis.fetch = async (...args) => { }; const sleep = ms => new Promise(resolve => setTimeout(resolve, ms)); const quantile = (a, q) => [...a].sort((x,y) => x-y)[Math.max(0, Math.ceil(a.length*q)-1)] ?? 0; -const summary = { origin, clients: count, seconds, captureMs, startedAt: new Date().toISOString(), synthetic: true, cleanup: [] }; +const summary = { origin, clients: count, seconds, captureMs, startedAt: new Date().toISOString(), synthetic: true, npubs: [], cleanup: [] }; try { for (let i = 0; i < count; i++) { const key = generateSecretKey(), identity = { nsec: nip19.nsecEncode(key), npub: nip19.npubEncode(getPublicKey(key)) }; const client = new HostedOnlyBotsClient({ apiBase: origin, ...identity }); + summary.npubs.push(identity.npub); clients.push(client); managers.push(new HotSyncManager({ stateDir: path.join(root, String(i)), origin, ...identity, profileName: 'launch-capacity-fixture' }, client)); } const start = performance.now(); let nextCapture = 0, event = 0; @@ -63,11 +64,14 @@ try { catch (error) { summary.cleanup.push({ device: i, scopeCleared: false, status: error.status, code: error.code }); } } const session = client.session(); - if (session) await originalFetch(`${origin}/v1/auth/revoke`, { method: 'POST', headers: { authorization: `Bearer ${session.sessionToken}` }, signal: AbortSignal.timeout(10000) }); + if (session) { + const revoked = await originalFetch(`${origin}/v1/auth/revoke`, { method: 'POST', headers: { authorization: `Bearer ${session.sessionToken}` }, signal: AbortSignal.timeout(10000) }); + summary.cleanup.push({device:i,sessionRevoked:revoked.ok,status:revoked.status}); + } } summary.requests = Object.values(observations.reduce((all, x) => { const key = `${x.phase} ${x.method} ${x.route}`; const row = all[key] ??= { route: key, count: 0, statuses: {}, latencies: [] }; row.count++; row.statuses[x.status] = (row.statuses[x.status] ?? 0)+1; row.latencies.push(x.ms); return all; }, {})).map(({latencies, ...row}) => ({ ...row, p95Ms: Math.round(quantile(latencies,.95)) })); summary.finishedAt = new Date().toISOString(); globalThis.fetch = originalFetch; rmSync(root, { recursive: true, force: true }); console.log(JSON.stringify(summary, null, 2)); } -if (summary.abort || summary.cleanup.some(x => !x.scopeCleared)) process.exitCode = 2; +if (summary.abort || summary.cleanup.some(x => x.scopeCleared === false || x.sessionRevoked === false)) process.exitCode = 2; diff --git a/scripts/smoke-coding-hosts.mjs b/scripts/smoke-coding-hosts.mjs index c4abd90..00973af 100644 --- a/scripts/smoke-coding-hosts.mjs +++ b/scripts/smoke-coding-hosts.mjs @@ -23,7 +23,7 @@ assert(nativeClaude && nativeCodex, 'Set CROSSHEAD_TEST_CLAUDE_BIN and CROSSHEAD // Construct a fresh environment instead of inheriting provider/config credentials. const env = { PATH: `${path.dirname(process.execPath)}:${process.env.PATH}`, TMPDIR: process.env.TMPDIR ?? os.tmpdir(), LANG: 'en_US.UTF-8', HOME: root, XDG_CONFIG_HOME: path.join(root, 'config'), XDG_DATA_HOME: path.join(root, 'data'), XDG_CACHE_HOME: path.join(root, 'cache'), - CROSSHEAD_HOME: home, ONLYBOTS_HOME: home, CROSSHEAD_HOT_SYNC_ENABLED: 'true', + CROSSHEAD_HOME: home, ONLYBOTS_HOME: home, CLAUDE_CONFIG_DIR: claudeHome, CLAUDE_CODE_HOME: claudeHome, CODEX_HOME: codexHome, CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: '1', DISABLE_AUTOUPDATER: '1', DISABLE_TELEMETRY: '1', ANTHROPIC_API_KEY: 'fixture-only', ANTHROPIC_AUTH_TOKEN: 'fixture-only', FIXTURE_API_KEY: 'fixture-only' }; diff --git a/src/runtime/cli.ts b/src/runtime/cli.ts index 444451d..e3684ed 100644 --- a/src/runtime/cli.ts +++ b/src/runtime/cli.ts @@ -1,4 +1,4 @@ -import { organizationJoinFromCli, signChallengeFromCli } from "./cli/organization"; +import { organizationJoinFromCli, signChallengeFromCli, signDeletionChallengeFromCli } from "./cli/organization"; import { installCodexFromCli, uninstallCodexFromCli } from "./codex-adapter"; import { installClaudeCodeFromCli, uninstallClaudeCodeFromCli } from "./claude-code-adapter"; import { installPiFromCli, uninstallPiFromCli } from "./pi-adapter"; @@ -158,6 +158,9 @@ export const runOnlyBotsCtlCli = async ( case "public": output = await publicFromCli(rest); break; + case "sign-deletion-challenge": + output = await signDeletionChallengeFromCli(rest); + break; case "sign-challenge": output = await signChallengeFromCli(rest); break; @@ -297,6 +300,7 @@ export const runOnlyBotsCtlCli = async ( "crosshead public offer --offer-id --task-ref --summary [--title ] [--description ] [--location ] [--listing-status ] [--published-at ] [--updated-at ] [--required-capability-ids ] [--required-capability-claim-refs ] [--target-agent-npubs ] [--collaboration-context-ref ] [--price --currency [--frequency ]] [--expires-at ] [--note ] [--home ] [--profile ] [--json|--debug]", "crosshead public claim --claim-id --offer-id --summary [--offer-event-id ] [--claim-state ] [--availability-ref ] [--collaboration-context-ref ] [--note ] [--home ] [--profile ] [--json|--debug]", "crosshead organization join [--agent-id ] [--profile ] [--json]", + 'crosshead sign-deletion-challenge "" --confirm erase-private-memory [--profile ] [--home ] [--json]', "crosshead sign-challenge \"\" [--npub ] [--profile ] [--home ] [--json]", "crosshead organization init --organization-id --organization-npub --display-name --summary --founder-agent-ids --founder-agent-npubs [--file ] [--json]", "crosshead organization inspect --file [--json]", diff --git a/src/runtime/cli/organization.ts b/src/runtime/cli/organization.ts index 1ad9816..b92dcff 100644 --- a/src/runtime/cli/organization.ts +++ b/src/runtime/cli/organization.ts @@ -51,3 +51,12 @@ export const signChallengeFromCli = async (argv: string[]): Promise => { } return [`Signed as ${result.npub} (profile ${result.profileName}). Paste this into the console:`, "", result.signature].join("\n"); }; + +/** Sign only the explicit irreversible private-memory erasure action. */ +export const signDeletionChallengeFromCli = async (argv: string[]): Promise => { + const {positionals,flags} = parseArgs(argv); + for (const flag of Object.keys(flags)) if (!["home","profile","json","confirm"].includes(flag)) throw new Error(`Unknown deletion-signing option: --${flag}`); + if (!positionals[0]) throw new Error('Usage: crosshead sign-deletion-challenge "" --confirm erase-private-memory [--profile name] [--home path] [--json]'); + const daemon = new OnlyBotsLocalRuntimeDaemon(typeof flags.home === "string" ? flags.home : undefined); + return JSON.stringify(daemon.signDeletionChallenge(positionals[0], {profileName:typeof flags.profile === "string" ? flags.profile : undefined,confirmation:typeof flags.confirm === "string" ? flags.confirm : undefined}),null,2); +}; diff --git a/src/runtime/erasure-proof.ts b/src/runtime/erasure-proof.ts new file mode 100644 index 0000000..da80955 --- /dev/null +++ b/src/runtime/erasure-proof.ts @@ -0,0 +1,23 @@ +import { createHash } from "node:crypto"; +import { schnorr } from "@noble/curves/secp256k1"; +import { nip19, getPublicKey } from "nostr-tools"; +const PURPOSE = "crosshead_erase_private_memory_v1"; +export type PrivateMemoryErasureChallenge = { purpose: typeof PURPOSE; origin: string; npub: string; nonce: string; issuedAt: string; expiresAt: string }; +export function parsePrivateMemoryErasureChallenge(text: string, now = new Date()): PrivateMemoryErasureChallenge { + if (text.length > 4096 || !text.startsWith(PURPOSE+":")) throw new Error("Expected a deletion-specific Crosshead challenge"); + const encoded = text.slice(PURPOSE.length+1); + if (!/^[A-Za-z0-9_-]+$/.test(encoded)) throw new Error("Invalid deletion challenge encoding"); + const c = JSON.parse(Buffer.from(encoded,"base64url").toString("utf8")) as PrivateMemoryErasureChallenge; + if (!c || Object.keys(c).sort().join(",") !== "expiresAt,issuedAt,nonce,npub,origin,purpose" || c.purpose !== PURPOSE || typeof c.origin !== "string" || new URL(c.origin).origin !== c.origin || !c.origin.startsWith("https://") || !/^[a-f0-9]{64}$/.test(c.nonce) || nip19.decode(c.npub).type !== "npub") throw new Error("Invalid deletion challenge"); + const issued = Date.parse(c.issuedAt), expires = Date.parse(c.expiresAt); + if (!Number.isFinite(issued) || !Number.isFinite(expires) || issued > now.getTime()+60000 || expires <= now.getTime() || expires <= issued || expires-issued > 86400000) throw new Error("Deletion challenge is expired or has invalid lifetime"); + return c; +} +export function signPrivateMemoryErasureChallenge(c: PrivateMemoryErasureChallenge, profile: { npub: string; apiBase: string }, nsec: string) { + if (c.npub !== profile.npub || c.origin !== new URL(profile.apiBase).origin) throw new Error("Deletion challenge does not match this profile's identity and server"); + const key = nip19.decode(nsec); if (key.type !== "nsec") throw new Error("Invalid local signing key"); + if (nip19.npubEncode(getPublicKey(key.data)) !== profile.npub) throw new Error("Local signing key does not match this profile"); + const message = JSON.stringify([c.purpose,c.origin,c.npub,c.nonce,c.issuedAt,c.expiresAt]); + const digest = createHash("sha256").update(message).digest(); + return { challenge:c, signature:Buffer.from(schnorr.sign(digest,key.data)).toString("hex") }; +} diff --git a/src/runtime/hot-sync-manager.ts b/src/runtime/hot-sync-manager.ts index bffdce8..bb1ebc6 100644 --- a/src/runtime/hot-sync-manager.ts +++ b/src/runtime/hot-sync-manager.ts @@ -8,7 +8,7 @@ import { hotSyncName, HOT_SYNC_PROTOCOL_VERSION, HOT_SYNC_MAX_SCOPES, type HotSy import { workspaceHotScope } from "./workspace-hot-memory"; import { hotSyncConflictIds, hotSyncSupersededIds } from "./hot-sync-signing"; import { normalizeTaskEvidence } from "./task-evidence"; -export const hotSyncEnabled = () => process.env.CROSSHEAD_HOT_SYNC_ENABLED === "true"; +export const hotSyncEnabled = () => process.env.CROSSHEAD_HOT_SYNC_ENABLED === undefined || process.env.CROSSHEAD_HOT_SYNC_ENABLED === "true"; type Config = { stateDir: string; origin: string; diff --git a/src/runtime/local-runtime.ts b/src/runtime/local-runtime.ts index 0a3bcf0..31baf11 100644 --- a/src/runtime/local-runtime.ts +++ b/src/runtime/local-runtime.ts @@ -1,3 +1,4 @@ +import { parsePrivateMemoryErasureChallenge, signPrivateMemoryErasureChallenge } from "./erasure-proof"; import { getHotSyncManager, hotSyncEnabled, stopHotSyncManagers } from "./hot-sync-manager"; import { shellQuote, writeManagedFile } from "./adapter-utils"; import { renderAgentMarkText } from "../protocol/agent-mark"; @@ -2581,6 +2582,17 @@ export class OnlyBotsLocalRuntimeDaemon { return { npub: profile.npub, signature, profileName: resolvedProfileName }; } + signDeletionChallenge(challenge: string, options: { profileName?: string; confirmation?: string } = {}) { + if (options.confirmation !== "erase-private-memory") throw new Error("Requires --confirm erase-private-memory. This authorizes hosted private-memory deletion and permanently disables this cloud identity; local files remain on your machine."); + const parsed = parsePrivateMemoryErasureChallenge(challenge); + const profileName = options.profileName ?? this.profileNameForNpub(parsed.npub) ?? loadRuntimeConfig(this.paths).defaultProfile; + if (!profileName) throw new Error("No matching runtime profile is configured"); + const {profile,secret} = this.loadProfileClient(profileName); + const proof = signPrivateMemoryErasureChallenge(parsed,profile,secret.nsec); + writeRuntimeLog(this.paths,"private_memory.erasure_proof_signed",{profileName}); + return proof; + } + async createBillingCheckout(request: Record, profileName?: string): Promise> { const { profileName: resolvedProfileName, secret, client, payload } = await this.withProfileBody({ profileName, diff --git a/tests/runtime/erasure-proof.test.ts b/tests/runtime/erasure-proof.test.ts new file mode 100644 index 0000000..c543a98 --- /dev/null +++ b/tests/runtime/erasure-proof.test.ts @@ -0,0 +1,12 @@ +import { describe, expect, it } from "vitest"; +import { createHash } from "node:crypto"; +import { schnorr } from "@noble/curves/secp256k1"; +import { generateSecretKey,getPublicKey,nip19 } from "nostr-tools"; +import { parsePrivateMemoryErasureChallenge,signPrivateMemoryErasureChallenge } from "../../src/runtime/erasure-proof"; +const origin="https://crosshead.sh"; +const fixture=()=>{const key=generateSecretKey(),npub=nip19.npubEncode(getPublicKey(key));return{key,npub,nsec:nip19.nsecEncode(key),challenge:{purpose:"crosshead_erase_private_memory_v1" as const,origin,npub,nonce:"7".repeat(64),issuedAt:new Date().toISOString(),expiresAt:new Date(Date.now()+60000).toISOString()}};}; +const encode=(c:unknown)=>"crosshead_erase_private_memory_v1:"+Buffer.from(JSON.stringify(c)).toString("base64url"); +describe("offline erasure signing",()=>{ + it("signs the domain-separated digest without exporting the private key",()=>{const f=fixture(),c=parsePrivateMemoryErasureChallenge(encode(f.challenge));const proof=signPrivateMemoryErasureChallenge(c,{npub:f.npub,apiBase:origin},f.nsec);const digest=createHash("sha256").update(JSON.stringify([c.purpose,c.origin,c.npub,c.nonce,c.issuedAt,c.expiresAt])).digest();expect(schnorr.verify(proof.signature,digest,getPublicKey(f.key))).toBe(true);expect(JSON.stringify(proof)).not.toContain(f.nsec);}); + it("rejects the wrong profile, wrong server, arbitrary signing payload and expired challenge",()=>{const f=fixture(),other=fixture();expect(()=>signPrivateMemoryErasureChallenge(f.challenge,{npub:other.npub,apiBase:origin},f.nsec)).toThrow();expect(()=>signPrivateMemoryErasureChallenge(f.challenge,{npub:f.npub,apiBase:"https://staging.crosshead.sh"},f.nsec)).toThrow();expect(()=>parsePrivateMemoryErasureChallenge("onlybots_link_"+"x".repeat(32))).toThrow();expect(()=>parsePrivateMemoryErasureChallenge(encode({...f.challenge,purpose:"crosshead_hosted_auth"}))).toThrow();expect(()=>parsePrivateMemoryErasureChallenge(encode({...f.challenge,expiresAt:new Date(Date.now()-1000).toISOString()}))).toThrow();}); +}); diff --git a/tests/runtime/hosted-lifecycle-security.test.ts b/tests/runtime/hosted-lifecycle-security.test.ts index 27b281e..b195e64 100644 --- a/tests/runtime/hosted-lifecycle-security.test.ts +++ b/tests/runtime/hosted-lifecycle-security.test.ts @@ -12,6 +12,7 @@ const homes: string[] = []; afterEach(() => { vi.restoreAllMocks(); + vi.unstubAllEnvs(); for (const home of homes.splice(0)) rmSync(home, { recursive: true, force: true }); }); @@ -138,6 +139,8 @@ describe("hosted lifecycle prompt boundary", () => { ["a successful hosted response", 200], ["a payment-required hosted response", 402], ] as const)("renders only client-verified hot memory after %s", async (_label, contextStatus) => { + // Exercise the explicit legacy opt-out path and its untrusted hosted responses. + vi.stubEnv("CROSSHEAD_HOT_SYNC_ENABLED", "false"); const signer = identity(); const home = mkdtempSync(path.join(os.tmpdir(), "crosshead-hosted-lifecycle-security-")); homes.push(home); diff --git a/tests/runtime/hot-sync-hook-path.test.ts b/tests/runtime/hot-sync-hook-path.test.ts index 9f51086..18a446b 100644 --- a/tests/runtime/hot-sync-hook-path.test.ts +++ b/tests/runtime/hot-sync-hook-path.test.ts @@ -7,7 +7,7 @@ import { generateSecretKey, getPublicKey, nip19 } from "nostr-tools"; import { HostRuntimeAdapterBase, type HostAdapterIdentity } from "../../src/runtime/host-adapter-base"; import { RuntimeLifecycleManager } from "../../src/runtime/lifecycle-engine"; import { HostedOnlyBotsClient } from "../../src/runtime/hosted-client"; -import { getHotSyncManager, stopHotSyncManagers } from "../../src/runtime/hot-sync-manager"; +import { getHotSyncManager, stopHotSyncManagers, hotSyncEnabled } from "../../src/runtime/hot-sync-manager"; import { buildContinuityGovernorState } from "../../src/runtime/continuity-governor"; import { buildDefaultContinuityPolicy } from "../../src/runtime/continuity-policy"; import type { LocalRuntimeClient } from "../../src/runtime/local-client"; @@ -53,6 +53,11 @@ afterEach(async () => { }); describe("real adapter local hot-memory path", () => { + it("uses local hot sync by default and supports an explicit rollback opt-out", () => { + vi.stubEnv("CROSSHEAD_HOT_SYNC_ENABLED", undefined); expect(hotSyncEnabled()).toBe(true); + vi.stubEnv("CROSSHEAD_HOT_SYNC_ENABLED", "false"); expect(hotSyncEnabled()).toBe(false); + vi.stubEnv("CROSSHEAD_HOT_SYNC_ENABLED", "true"); expect(hotSyncEnabled()).toBe(true); + }); it.each(["codex", "claude-code", "pi", "hermes", "opencode"])("prepares and durably completes %s while hosted requests hang", async runtime => { const h = harness(runtime); const input = { workspaceRoot: "/fixture-project", sessionId: "session", turnId: "first", prompt: "Repair the queue" }; From b1e89fcf428d0ed8f51e6af79a60000eb0f3ccbd Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 16:39:56 -0500 Subject: [PATCH 7/8] Add bounded production cross-machine sync acceptance --- .github/workflows/hot-sync-production.yml | 86 +++++++++++++++++++++++ scripts/accept-hot-sync-production.mjs | 74 +++++++++++++++++++ scripts/profile-hot-sync.mjs | 10 +-- 3 files changed, 166 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/hot-sync-production.yml create mode 100644 scripts/accept-hot-sync-production.mjs diff --git a/.github/workflows/hot-sync-production.yml b/.github/workflows/hot-sync-production.yml new file mode 100644 index 0000000..d905187 --- /dev/null +++ b/.github/workflows/hot-sync-production.yml @@ -0,0 +1,86 @@ +name: Production hot sync acceptance +on: + workflow_dispatch: +permissions: + contents: read +concurrency: + group: production-hot-sync-acceptance + cancel-in-progress: false +jobs: + linux-offline: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: {persist-credentials: false} + - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 + with: {node-version: 22, cache: npm} + - run: npm ci && npm run build + - name: Capture online and after a real Docker network disconnect + env: + CROSSHEAD_HN_FIXTURE_NSEC: ${{ secrets.CROSSHEAD_HN_20261106_NSEC }} + run: | + set -euo pipefail + mkdir -m 700 fixture-state + fixture_container="crosshead-hn-${GITHUB_RUN_ID}-a" + trap 'docker rm -f -v "$fixture_container" >/dev/null 2>&1 || true' EXIT + docker run -d --name "$fixture_container" --user "$(id -u):$(id -g)" --cpus 1 --memory 512m --memory-swap 512m --tmpfs /tmp:rw,size=64m --env CROSSHEAD_HN_FIXTURE_NSEC --mount "type=bind,src=$PWD,dst=/crosshead,readonly" --mount "type=bind,src=$PWD/fixture-state,dst=/fixture" node:22.23.4-bookworm-slim@sha256:de170b39439eb0e5a3444ad7d60b015614a381e10fecae2e30ec9e735122f38d sleep infinity + docker exec "$fixture_container" node /crosshead/scripts/accept-hot-sync-production.mjs --phase a-online --home /fixture/state --run-id "$GITHUB_RUN_ID" --allow-production + docker network disconnect bridge "$fixture_container" + docker exec "$fixture_container" node /crosshead/scripts/accept-hot-sync-production.mjs --phase a-offline --home /fixture/state --run-id "$GITHUB_RUN_ID" --allow-production + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 + with: {name: synthetic-linux-state, path: fixture-state/, retention-days: 7} + macos-recovery: + needs: linux-offline + runs-on: macos-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: {persist-credentials: false} + - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 + with: {node-version: 22, cache: npm} + - run: npm ci && npm run build + - name: Recover, reconnect, isolate and reset through production + env: + CROSSHEAD_HN_FIXTURE_NSEC: ${{ secrets.CROSSHEAD_HN_20261106_NSEC }} + run: node scripts/accept-hot-sync-production.mjs --phase b-reconnect-reset --home fixture-macos --run-id "$GITHUB_RUN_ID" --allow-production + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 + if: always() + with: {name: synthetic-macos-results, path: fixture-macos/result-*.json, retention-days: 7} + linux-reconnect: + needs: macos-recovery + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: {persist-credentials: false} + - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 + with: {node-version: 22, cache: npm} + - run: npm ci && npm run build + - uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 + with: {name: synthetic-linux-state, path: fixture-state} + - name: Reconnect stale offline state after the other machine reset + env: + CROSSHEAD_HN_FIXTURE_NSEC: ${{ secrets.CROSSHEAD_HN_20261106_NSEC }} + run: node scripts/accept-hot-sync-production.mjs --phase a-recover --home fixture-state/state --run-id "$GITHUB_RUN_ID" --allow-production + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 + if: always() + with: {name: synthetic-linux-recovery-results, path: fixture-state/state/result-a-recover.json, retention-days: 7} + cleanup: + needs: [linux-offline, macos-recovery, linux-reconnect] + if: always() + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: {persist-credentials: false} + - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 + with: {node-version: 22, cache: npm} + - run: npm ci && npm run build + - name: Clear purpose-owned production memory and revoke this session + env: + CROSSHEAD_HN_FIXTURE_NSEC: ${{ secrets.CROSSHEAD_HN_20261106_NSEC }} + run: node scripts/accept-hot-sync-production.mjs --phase cleanup --home fixture-cleanup --run-id "$GITHUB_RUN_ID" --allow-production + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 + if: always() + with: {name: synthetic-cleanup-result, path: fixture-cleanup/result-cleanup.json, retention-days: 7} diff --git a/scripts/accept-hot-sync-production.mjs b/scripts/accept-hot-sync-production.mjs new file mode 100644 index 0000000..4773959 --- /dev/null +++ b/scripts/accept-hot-sync-production.mjs @@ -0,0 +1,74 @@ +#!/usr/bin/env node +// Disposable production customer identity only. State artifacts contain signed +// synthetic memory, never the private key or bearer tokens. +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { mkdirSync, readFileSync, readdirSync, writeFileSync, chmodSync } from 'node:fs'; +import path from 'node:path'; +const require=createRequire(import.meta.url); +const {HotSyncManager}=require('../dist/runtime/hot-sync-manager.js'); +const {HostedOnlyBotsClient}=require('../dist/runtime/hosted-client.js'); +const {generateSecretKey,getPublicKey,nip19}=require('nostr-tools'); +const option=(name,fallback)=>{const i=process.argv.indexOf(name);return i<0?fallback:process.argv[i+1];}; +const phase=option('--phase'),home=path.resolve(option('--home','fixture-state')),run=option('--run-id'); +const origin='https://crosshead.sh',profileName=`hn-fixture-${run}`,workspace='repo:/synthetic-crosshead-project'; +assert(run&&process.argv.includes('--allow-production'),'Explicit run ID and --allow-production required'); +const nsec=process.env.CROSSHEAD_HN_FIXTURE_NSEC;assert(nsec,'Missing purpose-owned synthetic identity'); +const decoded=nip19.decode(nsec);assert.equal(decoded.type,'nsec'); +const npub=nip19.npubEncode(getPublicKey(decoded.data)); +mkdirSync(home,{recursive:true,mode:0o700}); +const privatize=dir=>{chmodSync(dir,0o700);for(const e of readdirSync(dir,{withFileTypes:true})){const p=path.join(dir,e.name);assert(!e.isSymbolicLink(),'Fixture artifacts must not contain symlinks');if(e.isDirectory())privatize(p);else chmodSync(p,0o600);}};privatize(home); +const originalFetch=globalThis.fetch;let offline=false; +globalThis.fetch=(...args)=>offline?Promise.reject(new Error('fixture offline transport')):originalFetch(...args); +const clients=[],managers=[],checks=[]; +const make=(stateDir=home,key=nsec,name=profileName)=>{ + const d=nip19.decode(key),pub=nip19.npubEncode(getPublicKey(d.data)),client=new HostedOnlyBotsClient({apiBase:origin,npub:pub,nsec:key}); + const m=new HotSyncManager({stateDir,origin,npub:pub,nsec:key,profileName:name},client);m.link('/synthetic-crosshead-project',`hn-fixture-${run}`);clients.push(client);managers.push(m);return m; +}; +const findings=m=>[...new Map(m.read().scopes.flatMap(s=>JSON.parse(s.content).continuityCheckpoints).map(c=>[c.id,c])).values()]; +const capture=(m,id)=>{const begin=performance.now();m.capture('codex',{id:`${run}:${id}`,runtime:'codex',workspaceId:workspace,occurredAt:new Date().toISOString(),goal:'Synthetic cross-machine launch acceptance',summary:`Synthetic ${id}; canonical duration zero is valid, negatives rejected.`});return performance.now()-begin;}; +const has=(m,id)=>findings(m).some(c=>c.id===`${run}:${id}`); +const sync=async(m,predicate)=>{const deadline=Date.now()+90000;let last;do{try{await m.flush(false);if(predicate())return;}catch(error){last=error;}await new Promise(r=>setTimeout(r,1500));}while(Date.now()m.status().scopes.find(s=>s.scope.startsWith('ws-codex-'))?.scope; +let m, primaryFailure=null; +try{ + m=make(); + if(phase==='a-online'){ + capture(m,'a-base');await sync(m,()=>m.status().pendingCount===0);assert(has(m,'a-base'));checks.push('Linux A uploaded signed synthetic finding'); + }else if(phase==='a-offline'){ + assert(has(m,'a-base'));const ms=capture(m,'a-stale');m.tick();const failedBy=Date.now()+30000;while(!m.status().error&&Date.now()setTimeout(r,100));await m.stop();assert.equal(m.status().error,'sync_unavailable');assert.equal(m.status().pendingCount,1);assert(has(m,'a-stale'));checks.push('Docker network disconnected; capture fsynced and remained pending');checks.push(`offline capture ${ms.toFixed(2)} ms`); + }else if(phase==='b-reconnect-reset'){ + await sync(m,()=>has(m,'a-base'));checks.push('Fresh macOS B recovered Linux A finding from production'); + capture(m,'b-online');await sync(m,()=>m.status().pendingCount===0); + offline=true;capture(m,'b-offline');await m.stop();m=make();assert(has(m,'b-offline'));assert.equal(m.status().pendingCount,1);checks.push('macOS offline transport capture survived restart'); + offline=false;await sync(m,()=>m.status().pendingCount===0); + const fresh=make(path.join(home,'fresh-recovery'));await sync(fresh,()=>has(fresh,'b-offline')&&has(fresh,'a-base')); + const other=make(path.join(home,'other-identity'),nip19.nsecEncode(generateSecretKey()));await sync(other,()=>!other.status().initialSyncPending);assert.equal(findings(other).length,0); + const otherProfile=make(path.join(home,'other-profile'),nsec,`${profileName}-other`);await sync(otherProfile,()=>!otherProfile.status().initialSyncPending);assert.equal(findings(otherProfile).length,0);checks.push('Fresh recovery and identity/profile isolation verified'); + m.reset(scope(m));await sync(m,()=>!m.status().scopes.some(s=>s.pendingReset));assert.equal(findings(m).length,0);checks.push('macOS reset reached production'); + }else if(phase==='a-recover'){ + assert(has(m,'a-stale'));assert.equal(m.status().pendingCount,1); + await sync(m,()=>m.status().pendingCount===0);assert(!has(m,'a-stale'));assert.equal(findings(m).length,0);assert(m.status().scopes.some(s=>s.epoch>0&&s.discardedByReset>0));checks.push('Linux stale offline capture discarded after remote reset, without resurrection'); + capture(m,'after-reset');await sync(m,()=>m.status().pendingCount===0); + const fresh=make(path.join(home,'fresh-after-reset'));await sync(fresh,()=>has(fresh,'after-reset'));assert(!has(fresh,'a-stale'));checks.push('New epoch accepts new work and fresh recovery'); + }else if(phase==='cleanup'){ + await sync(m,()=>!m.status().initialSyncPending); + for(const row of m.status().scopes){m.reset(row.scope);await sync(m,()=>!m.status().scopes.some(s=>s.pendingReset));} + assert.equal(findings(m).length,0);checks.push('All purpose-owned production scopes cleared'); + }else throw new Error('Unknown acceptance phase'); + const result={phase,platform:process.platform,run,npub,checks,pending:m.status().pendingCount,scopes:m.status().scopes.map(s=>({scope:s.scope,epoch:s.epoch,revision:s.remoteRevision,discarded:s.discardedByReset}))}; + writeFileSync(path.join(home,`result-${phase}.json`),JSON.stringify(result,null,2)+'\n');console.log(JSON.stringify(result)); +}catch(error){primaryFailure=error;throw error;}finally{ + const failures=[];offline=false; + const stopped=await Promise.allSettled(managers.map(m=>m.stop()));for(const result of stopped)if(result.status==='rejected')failures.push('manager stop failed'); + for(const [index,client] of clients.entries()){ + const session=client.session();if(!session)continue; + try{const response=await originalFetch(`${origin}/v1/auth/revoke`,{method:'POST',headers:{authorization:`Bearer ${session.sessionToken}`},signal:AbortSignal.timeout(10000)});const body=await response.json();assert(response.ok&&body.revoked===true);} + catch{failures.push(`session revocation failed for fixture client ${index}`);} + } + globalThis.fetch=originalFetch; + // Enforce the artifact contract even if transport cleanup failed. + const inspect=dir=>{for(const e of readdirSync(dir,{withFileTypes:true})){const file=path.join(dir,e.name);if(e.isDirectory())inspect(file);else{const content=readFileSync(file,'utf8');assert(!content.includes(nsec)&&!content.includes('nsec1')&&!content.includes('ob_sess_'),'Secret found in fixture artifact');}}}; + try{inspect(home);}catch(error){failures.push(error.message);} + if(failures.length){writeFileSync(path.join(home,'cleanup-errors.json'),JSON.stringify({failures},null,2)+'\n');if(!primaryFailure)throw new Error(failures.join('; '));} +} diff --git a/scripts/profile-hot-sync.mjs b/scripts/profile-hot-sync.mjs index a505e92..5c089d4 100644 --- a/scripts/profile-hot-sync.mjs +++ b/scripts/profile-hot-sync.mjs @@ -56,7 +56,7 @@ try { summary.devices = managers.map(m => { const s = m.status(); return { state: s.state, pending: s.pendingCount, error: s.error }; }); } finally { phase = 'cleanup'; - await Promise.all(managers.map(m => m.stop())); + for (const [device,result] of (await Promise.allSettled(managers.map(m => m.stop()))).entries()) if(result.status === "rejected") summary.cleanup.push({device,stopFailed:true}); for (const [i, manager] of managers.entries()) { const client = clients[i]; for (const {scope} of manager.status().scopes) { @@ -65,8 +65,10 @@ try { } const session = client.session(); if (session) { - const revoked = await originalFetch(`${origin}/v1/auth/revoke`, { method: 'POST', headers: { authorization: `Bearer ${session.sessionToken}` }, signal: AbortSignal.timeout(10000) }); - summary.cleanup.push({device:i,sessionRevoked:revoked.ok,status:revoked.status}); + try { const revoked = await originalFetch(`${origin}/v1/auth/revoke`, { method: 'POST', headers: { authorization: `Bearer ${session.sessionToken}` }, signal: AbortSignal.timeout(10000) }); + const result = await revoked.json(); + summary.cleanup.push({device:i,sessionRevoked:revoked.ok && result.revoked === true,status:revoked.status}); + } catch { summary.cleanup.push({device:i,sessionRevoked:false}); } } } summary.requests = Object.values(observations.reduce((all, x) => { const key = `${x.phase} ${x.method} ${x.route}`; const row = all[key] ??= { route: key, count: 0, statuses: {}, latencies: [] }; row.count++; row.statuses[x.status] = (row.statuses[x.status] ?? 0)+1; row.latencies.push(x.ms); return all; }, {})).map(({latencies, ...row}) => ({ ...row, p95Ms: Math.round(quantile(latencies,.95)) })); @@ -74,4 +76,4 @@ try { globalThis.fetch = originalFetch; rmSync(root, { recursive: true, force: true }); console.log(JSON.stringify(summary, null, 2)); } -if (summary.abort || summary.cleanup.some(x => x.scopeCleared === false || x.sessionRevoked === false)) process.exitCode = 2; +if (summary.abort || summary.cleanup.some(x => x.scopeCleared === false || x.sessionRevoked === false || x.stopFailed)) process.exitCode = 2; From de1a520bf7166d36d74937eab0b3e2982213868d Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 16:40:32 -0500 Subject: [PATCH 8/8] Report and verify production fixture teardown --- .github/workflows/hot-sync-production.yml | 15 ++++++++++++++- scripts/accept-hot-sync-production.mjs | 2 +- scripts/profile-hot-sync.mjs | 2 +- 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/.github/workflows/hot-sync-production.yml b/.github/workflows/hot-sync-production.yml index d905187..e52ca90 100644 --- a/.github/workflows/hot-sync-production.yml +++ b/.github/workflows/hot-sync-production.yml @@ -23,7 +23,20 @@ jobs: set -euo pipefail mkdir -m 700 fixture-state fixture_container="crosshead-hn-${GITHUB_RUN_ID}-a" - trap 'docker rm -f -v "$fixture_container" >/dev/null 2>&1 || true' EXIT + cleanup() { + result=$? + trap - EXIT + if ! docker rm -f -v "$fixture_container" >/dev/null; then + echo 'Owned fixture container removal failed' >&2 + exit 1 + fi + if docker inspect "$fixture_container" >/dev/null 2>&1; then + echo 'Owned fixture container still exists' >&2 + exit 1 + fi + exit "$result" + } + trap cleanup EXIT docker run -d --name "$fixture_container" --user "$(id -u):$(id -g)" --cpus 1 --memory 512m --memory-swap 512m --tmpfs /tmp:rw,size=64m --env CROSSHEAD_HN_FIXTURE_NSEC --mount "type=bind,src=$PWD,dst=/crosshead,readonly" --mount "type=bind,src=$PWD/fixture-state,dst=/fixture" node:22.23.4-bookworm-slim@sha256:de170b39439eb0e5a3444ad7d60b015614a381e10fecae2e30ec9e735122f38d sleep infinity docker exec "$fixture_container" node /crosshead/scripts/accept-hot-sync-production.mjs --phase a-online --home /fixture/state --run-id "$GITHUB_RUN_ID" --allow-production docker network disconnect bridge "$fixture_container" diff --git a/scripts/accept-hot-sync-production.mjs b/scripts/accept-hot-sync-production.mjs index 4773959..ebfc91c 100644 --- a/scripts/accept-hot-sync-production.mjs +++ b/scripts/accept-hot-sync-production.mjs @@ -70,5 +70,5 @@ try{ // Enforce the artifact contract even if transport cleanup failed. const inspect=dir=>{for(const e of readdirSync(dir,{withFileTypes:true})){const file=path.join(dir,e.name);if(e.isDirectory())inspect(file);else{const content=readFileSync(file,'utf8');assert(!content.includes(nsec)&&!content.includes('nsec1')&&!content.includes('ob_sess_'),'Secret found in fixture artifact');}}}; try{inspect(home);}catch(error){failures.push(error.message);} - if(failures.length){writeFileSync(path.join(home,'cleanup-errors.json'),JSON.stringify({failures},null,2)+'\n');if(!primaryFailure)throw new Error(failures.join('; '));} + if(failures.length){console.error('Fixture cleanup failures: '+failures.join('; '));writeFileSync(path.join(home,'cleanup-errors.json'),JSON.stringify({failures},null,2)+'\n');if(!primaryFailure)throw new Error(failures.join('; '));} } diff --git a/scripts/profile-hot-sync.mjs b/scripts/profile-hot-sync.mjs index 5c089d4..c34f012 100644 --- a/scripts/profile-hot-sync.mjs +++ b/scripts/profile-hot-sync.mjs @@ -68,7 +68,7 @@ try { try { const revoked = await originalFetch(`${origin}/v1/auth/revoke`, { method: 'POST', headers: { authorization: `Bearer ${session.sessionToken}` }, signal: AbortSignal.timeout(10000) }); const result = await revoked.json(); summary.cleanup.push({device:i,sessionRevoked:revoked.ok && result.revoked === true,status:revoked.status}); - } catch { summary.cleanup.push({device:i,sessionRevoked:false}); } + } catch { summary.cleanup.push({device:i,sessionRevoked:false}); } } } summary.requests = Object.values(observations.reduce((all, x) => { const key = `${x.phase} ${x.method} ${x.route}`; const row = all[key] ??= { route: key, count: 0, statuses: {}, latencies: [] }; row.count++; row.statuses[x.status] = (row.statuses[x.status] ?? 0)+1; row.latencies.push(x.ms); return all; }, {})).map(({latencies, ...row}) => ({ ...row, p95Ms: Math.round(quantile(latencies,.95)) }));