diff --git a/.github/workflows/coding-hosts.yml b/.github/workflows/coding-hosts.yml new file mode 100644 index 0000000..1fc8d00 --- /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 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 + 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/.github/workflows/hot-sync-production.yml b/.github/workflows/hot-sync-production.yml new file mode 100644 index 0000000..e52ca90 --- /dev/null +++ b/.github/workflows/hot-sync-production.yml @@ -0,0 +1,99 @@ +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" + 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" + 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/.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..837bec2 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 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 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. @@ -140,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/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/accept-hot-sync-production.mjs b/scripts/accept-hot-sync-production.mjs new file mode 100644 index 0000000..ebfc91c --- /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){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 new file mode 100644 index 0000000..c34f012 --- /dev/null +++ b/scripts/profile-hot-sync.mjs @@ -0,0 +1,79 @@ +#!/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, 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; + 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'; + 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) { + 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) { + 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)) })); + 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 === false || x.sessionRevoked === false || x.stopFailed)) process.exitCode = 2; diff --git a/scripts/smoke-coding-hosts.mjs b/scripts/smoke-coding-hosts.mjs new file mode 100644 index 0000000..00973af --- /dev/null +++ b/scripts/smoke-coding-hosts.mjs @@ -0,0 +1,140 @@ +#!/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, 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 = 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'); +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, + 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 }; +let primaryFailure = null; +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)); +} 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'); + 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 }); + } +} 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 8441bbe..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; @@ -22,6 +22,7 @@ type Metadata = { linked: Record; migrated: Record; mixedVersion: boolean; + recoverySnapshots?: string[]; lastDiscoveryAt?: string; error?: string | null; }; @@ -49,7 +50,12 @@ 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 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"); @@ -82,6 +88,8 @@ export class HotSyncManager { catch { } const id = this.digest(remote ? `git:${remote}` : `local:${raw}`); this.metadata.workspaces[raw] = id; + this.nextLegacyCheck = 0; + this.workspaceGeneration++; this.persist(); return id; } @@ -94,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) }; } @@ -138,22 +148,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 +224,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 +249,15 @@ 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) { + const generation = this.workspaceGeneration; + await this.migrateLegacy(); + this.nextLegacyCheck = generation === this.workspaceGeneration ? Date.now() + 60000 : 0; + } + 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 +270,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 +277,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; @@ -252,12 +317,15 @@ 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 : []) { 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/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-acceptance.test.ts b/tests/runtime/hot-sync-acceptance.test.ts index 7729a67..267d651 100644 --- a/tests/runtime/hot-sync-acceptance.test.ts +++ b/tests/runtime/hot-sync-acceptance.test.ts @@ -54,6 +54,110 @@ 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("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("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" }; + 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..18a446b --- /dev/null +++ b/tests/runtime/hot-sync-hook-path.test.ts @@ -0,0 +1,174 @@ +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, 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"; + +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("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" }; + 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(); } + }); + +});