diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e4068b0..1f210db 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -186,6 +186,8 @@ jobs: runner: ubuntu-24.04 - platform: linux-arm64 runner: ubuntu-24.04-arm + - platform: darwin-arm64 + runner: macos-15 name: official-omp-gate0 (${{ matrix.platform }}) runs-on: ${{ matrix.runner }} timeout-minutes: 10 @@ -210,15 +212,23 @@ jobs: with: bun-version: 1.3.14 + - name: Install dependencies + run: pnpm install --frozen-lockfile + - name: Prove the pinned unmodified official OMP runtime run: pnpm --filter @t4-code/host-service verify:official-omp-lifecycle + - name: Prove the packaged T4 host against official OMP + run: pnpm --filter @t4-code/host-daemon verify:official-omp-packaged + - name: Upload official OMP Gate 0 evidence if: ${{ success() }} uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 with: name: official-omp-gate0-${{ matrix.platform }}-${{ github.run_id }} - path: artifacts/official-omp-gate0/${{ matrix.platform }}.json + path: | + artifacts/official-omp-gate0/${{ matrix.platform }}.json + artifacts/official-omp-packaged-host/${{ matrix.platform }}.json if-no-files-found: error retention-days: 14 diff --git a/compat/official-omp-gate0.json b/compat/official-omp-gate0.json index e892515..3968ea0 100644 --- a/compat/official-omp-gate0.json +++ b/compat/official-omp-gate0.json @@ -29,14 +29,30 @@ "liveSessionEntries": true, "durableCommandKey": true }, + "t4AdapterCoverage": { + "jsonlTranscriptReconciliation": true, + "synthesizedReadyWatermark": true, + "durableEntryProjection": true, + "liveEntryDeduplication": true, + "conservativePromptCorrelation": true, + "durableCommandKey": false + }, + "packagedHostProof": { + "requiredPlatforms": ["darwin-arm64", "linux-x64", "linux-arm64"], + "requiredScenarios": ["discovery", "attach", "prompt", "durable-jsonl", "t4-wire-projection"], + "authorityMode": "official-exclusive-profile", + "releasedDefault": "lycaon-authority-bridge" + }, "t4Policy": { "transcriptAuthority": "official-omp-jsonl", "ambiguousDispatch": "outcome-unknown-no-auto-replay", - "releasedFallback": "lycaon-authority-bridge-until-packaged-cutover" + "releasedFallback": "lycaon-authority-bridge-until-local-and-managed-cutover" }, "evidence": { "runtimeHarness": "packages/host-service/bin/official-omp-gate0.ts", + "packagedHostHarness": "packages/host-daemon/bin/official-omp-packaged-proof.ts", "platformOutput": "artifacts/official-omp-gate0/-.json", + "packagedPlatformOutput": "artifacts/official-omp-packaged-host/-.json", "dispatchCrashTest": "packages/host-service/test/hardening.test.ts", "clientNoReplayTest": "packages/client/test/client.test.ts" } diff --git a/docs/OMP_T4_CAPABILITY_AUDIT.md b/docs/OMP_T4_CAPABILITY_AUDIT.md index 6ee1c84..7e7b52a 100644 --- a/docs/OMP_T4_CAPABILITY_AUDIT.md +++ b/docs/OMP_T4_CAPABILITY_AUDIT.md @@ -14,7 +14,7 @@ The largest problem is no longer a missing transport layer. T4 now has a shared 1. **Operation truth is implemented in the host but was not yet reaching the UI.** PR #111 added `typed`, `headless`, `terminal-only`, and `unavailable`; PR #113 classifies stock OMP commands and rejects known terminal-only text before it reaches the model. This sprint carries that result into desktop/web and Flutter. 2. **Important daily workflows still lack a typed app action:** plan and goal modes, session branch/fork/tree, handoff, provider login/logout, queue control, and child-agent steering. -3. **The official-OMP Gate 0 proof is complete.** Unmodified OMP 17.0.6 passed restart continuity, steering, follow-up, approval, cancellation, and the no-replay failure policy on macOS ARM64 plus native Linux x64 and ARM64. +3. **The official-OMP Gate 0 proof is complete, and the first packaged-host proof passes.** Unmodified OMP 17.0.6 passed restart continuity, steering, follow-up, approval, cancellation, and the no-replay failure policy on macOS ARM64 plus native Linux x64 and ARM64. The compiled `t4-host` now also passes discovery, attach, prompt, durable JSONL, and T4-wire projection against official OMP on macOS ARM64; the same packaged proof is wired into the native Linux jobs. 4. **The Lycaon fork is transitional, not the desired product center.** The current public package still pins its thin authority bridge, but new compatibility work should prefer the T4-owned official-OMP adapter and add only small, extractable bridge methods when stock OMP has no usable seam. 5. **Protocol vocabulary is not UI coverage.** Every tracker row needs an explicit client disposition: direct control, palette action, read-only view, disabled explanation, terminal handoff, or unavailable. 6. **Source is ahead of the public release.** Flutter and the latest adapter work are on `main`; the latest public GitHub release remains v0.1.28. @@ -29,7 +29,9 @@ The truthful-command foundation is split cleanly between merged host work and th - **Merged:** PR #117 preserves `catalog.get.result.operations` through the desktop runtime. - **Merged:** PRs #118 and #120 make web/Electron and Flutter build truthful slash menus from the runtime capability contract and fail closed when that contract is absent or unavailable. - **This sprint:** the official OMP 17.0.6 harness passes lifecycle, crash/resume, steer, follow-up, approval, and cancellation on macOS ARM64; the host/client failure tests prove a dispatched command becomes `outcome_unknown` and is never automatically replayed. -- **This sprint:** `compat/official-omp-gate0.json` records the exact runtime, required platforms/scenarios, direct RPC support, and three missing stock seams. Linux x64 and ARM64 CI jobs pass and retain per-platform evidence. +- **This sprint:** `compat/official-omp-gate0.json` records the exact runtime, required platforms/scenarios, direct RPC support, and three missing stock seams. macOS ARM64 plus Linux x64 and ARM64 CI jobs retain per-platform lifecycle and packaged-host evidence. +- **This sprint:** the shared supervisor synthesizes the missing ready watermark from official JSONL, projects complete durable records, conservatively correlates matching user prompts, and deduplicates the Lycaon fallback's richer live frames. +- **This sprint:** `t4-host --omp-authority official --omp-sessions-root ` adds an opt-in, exclusive-profile cutover path. The compiled macOS ARM64 host passes the real Unix-socket proof; native Linux proof is now a required CI scenario. The released/default desktop path remains the Lycaon bridge. - **This sprint:** Flutter adds project Quick Open plus visible pause, resume, and manual compaction controls over existing typed commands. No new fork behavior is required. The tracker distinguishes merged source, work in this sprint, and public release state. None of the new adapter/client work is claimed as packaged desktop, Android, or iOS proof yet. @@ -40,7 +42,7 @@ The tracker distinguishes merged source, work in this sprint, and public release |---|---|---|---|---|---| | Operation capability contract | Merged | Decodes shared contract | Same web client | Dart decoder merged | PR #111 green | | Official OMP command discovery and rejection | Merged | Receives through host | Receives through host | Receives through host | PR #113 green | -| Official OMP Gate 0 behavior | macOS ARM64 and native Linux x64/ARM64 pass | Shared host behavior | Shared host behavior | Shared host behavior | Stock 17.0.6 passes lifecycle, steer/follow-up, approval, cancellation, and crash/no-replay behavior | +| Official OMP Gate 0 behavior | Direct RPC passes macOS ARM64 and native Linux x64/ARM64; compiled local host passes macOS ARM64 and is required in Linux CI | Adapter path proven, packaged desktop default unchanged | Shared host behavior | Shared host behavior | Stock 17.0.6 passes lifecycle, steer/follow-up, approval, cancellation, crash/no-replay, JSONL reconciliation, and the local T4 wire | | Preserve `catalog.get.operations` | Merged host response | Merged in PR #117 | Merged in PR #117 | Already decoded | Response and live-frame client tests pass | | Capability-aware slash menu | Host rejects unsafe fallback | Implemented on `main` | Implemented on `main` | Implemented on `main` | PRs #118 and #120 are green and merged | | Project Quick Open | `files.search` merged | Implemented on `main` | Implemented on `main` | Implemented this sprint | Flutter analysis and the full 168-test suite pass locally; platform CI pending | @@ -60,7 +62,7 @@ This matrix is intentionally stricter than “the protocol supports it.” A row | Flutter merge | [`LycaonLLC/t4-code#104`](https://github.com/LycaonLLC/t4-code/pull/104) | New shared desktop/mobile client now on `main` | | Public T4 release | [`v0.1.28`](https://github.com/LycaonLLC/t4-code/releases/tag/v0.1.28) | Latest public release visible during the audit | -This is primarily a source audit, supplemented by merged official-adapter smokes and focused TypeScript tests in this sprint. It does not prove that every path works in a packaged app. Live Flutter desktop, Android, iOS, web, and Electron round trips remain a separate verification pass. +This is primarily a source audit, supplemented by merged official-adapter smokes, focused TypeScript tests, and one compiled-host proof. It does not prove that every path works in a packaged desktop or mobile app. Live Flutter desktop, Android, iOS, web, and Electron round trips remain a separate verification pass. Relevant planning and implementation changes: @@ -152,7 +154,8 @@ Fork PR [#22](https://github.com/lyc-aon/oh-my-pi/pull/22) removed more than 37, | Priority | Gap | Why it matters | Best patch path | |---|---|---|---| -| T0 | Packaged official-OMP cutover is not proven | Gate 0 now passes on macOS ARM64 and native Linux x64/ARM64, but released builds still use the Lycaon fallback | Reconcile stock JSONL through the shared adapter, then run packaged local and managed cutover proofs | +| T0 | Managed and shipped official-OMP cutover is not proven | Stock JSONL reconciliation and the compiled local host now pass on macOS ARM64; Linux packaged-host CI, managed-session proof, and an actual desktop release still remain | Land native packaged-host evidence, prove the same authority in managed sessions, then change release packaging deliberately | +| T0 | A fresh official-OMP profile has no project bootstrap flow | The current authority derives projects from existing sessions, so the opt-in proof seeds one session before discovery | Add a small T4-owned project registry or explicit project-add command, then prove first-run create without prewriting OMP state | | T0 | Release state is ambiguous | Source says v0.1.30 while public GitHub release remains v0.1.28 | Separate `on main`, `verified package`, and `publicly released` in the tracker/release gate | | T0 | Plan, goal, branch/fork/tree, handoff, and provider auth lack complete typed app flows | These are central OMP workflows, not decorative terminal features | Typed T4 commands backed by existing OMP RPC where possible | | T1 | Queue and pause/resume controls are not consistently exposed outside Flutter | Cross-device control needs explicit, predictable behavior | Add equally visible controls to web/Capacitor over the same typed commands | @@ -252,7 +255,7 @@ Do not send a giant T4 product PR to original OMP. A suitable upstream contribut ## Capability manifest and drift control -The checked snapshot at `compat/official-omp-gate0.json` now records the Gate 0 runtime, platform/scenario contract, direct RPC support, missing stock seams, and no-replay policy. Each native run writes `artifacts/official-omp-gate0/-.json` as evidence. The broader capability snapshot should next add: +The checked snapshot at `compat/official-omp-gate0.json` now records the Gate 0 runtime, platform/scenario contract, direct RPC support, missing stock seams, T4's JSONL coverage, packaged-host contract, and no-replay policy. Native runs write both `artifacts/official-omp-gate0/-.json` and `artifacts/official-omp-packaged-host/-.json` as evidence. The broader capability snapshot should next add: - Official OMP version and exact commit; when the fallback bridge is exercised, its fork commit and tag too. - Operations and aliases using `typed`, `headless`, `terminal-only`, or `unavailable`. @@ -279,9 +282,10 @@ official OMP version + commit ### Phase 0: finish the official-OMP foundation -1. Reconcile stock JSONL without depending on fork-only ready watermarks or live `session_entry` frames. -2. Run packaged local and managed cutover proofs while preserving the released fallback. -3. Track source, verified package, and public release status separately, then retire fallback use only after those proofs pass. +1. Land the completed stock-JSONL reconciliation and native packaged-local proof without changing the released default. +2. Add a first-run project registry or explicit project-add flow so an empty exclusive profile can create its first session. +3. Apply the same exclusive-profile authority contract to managed session hosts and prove it on native Linux x64/ARM64. +4. Track source, verified package, and public release status separately, then retire fallback use only after packaged desktop and managed proofs pass. ### Phase 1: close the daily-workflow gaps @@ -320,7 +324,7 @@ For each future sprint: ## Audit limitations and next proof pass -- This pass did not run packaged OMP or T4 binaries. +- This pass ran the compiled `t4-host` against the pinned official OMP binary on macOS ARM64, but did not run a packaged Electron/Flutter application bundle. - It did not visually inspect every desktop, Android, or iOS screen. - Optional protocol features may not be negotiated by every host instance. - The shared adapter is currently an official-OMP seam, not proof that T4 is a generic multi-runtime product. diff --git a/docs/OMP_T4_CAPABILITY_TRACKER.csv b/docs/OMP_T4_CAPABILITY_TRACKER.csv index 365749b..a0bcf4e 100644 --- a/docs/OMP_T4_CAPABILITY_TRACKER.csv +++ b/docs/OMP_T4_CAPABILITY_TRACKER.csv @@ -1,13 +1,13 @@ "ID","Tier","Area","Capability","Original OMP 17.0.6","Lycaon t4code/main","Desktop web/Electron","Mobile web/Capacitor","Flutter main","Gap classification","Recommended route","Evidence note" "H01","T0","Host","Protocol and capability negotiation","RPC and ACP have typed requests; RPC has no explicit version field","Adds versioned omp-app/1 and t4-omp-authority/1","Code: full","Code: full","Code: full","No product gap; upstream RPC versioning would reduce adapter ambiguity","Keep T4 wire; consider a small generic upstream RPC version field","Clients must honor the features and permissions in the live welcome frame" -"H02","T0","Host","Local host discovery and lifecycle","CLI runtime","Adds T4 host compatibility launcher and OMP authority","Code: full","N/A: remote gateway client","Code: partial","Flutter cutover and packaged lifecycle proof remain","Platform adapter over T4-owned host lifecycle","Do not move generic lifecycle back into OMP" +"H02","T0","Host","Local host discovery and lifecycle","CLI runtime","Adds T4 host compatibility launcher and OMP authority","Code: partial","N/A: remote gateway client","Code: partial","Compiled t4-host passes an opt-in official-OMP exclusive-profile lifecycle proof on macOS ARM64, but an empty profile cannot register its first project without a seeded session; packaged app and Linux CI evidence remain","Add a T4-owned project registry or project-add flow; keep OMP authoritative for session execution","The Lycaon bridge remains the released default" "H03","T0","Host","Remote pairing and multiple hosts","Not an OMP product concern","T4 host pairing authority","Code: full","Code: full","Code: full","Needs packaged cross-platform proof","Client UI over existing T4 protocol","T4-owned feature; not an upstream parity item" -"H04","T0","Host","Reconnect replay and transcript reconciliation","RPC stream and session persistence","T4 host adds replay and durable projection","Code: full","Code: full","Code: full","Needs live interruption tests per platform","Verify existing protocol and clients","Protocol vocabulary is present" +"H04","T0","Host","Reconnect replay and transcript reconciliation","RPC stream and session persistence","T4 host adds replay and durable projection","Code: full","Code: full","Code: full","Official JSONL reconciliation now synthesizes watermarks, ignores partial records, streams large transcripts, and deduplicates fallback live frames; client interruption proof remains","Verify packaged clients over the shared adapter","Host unit and compiled local-host proofs pass" "H05","T1","Host","Offline cached transcript","Persistent local sessions","T4 clients cache projected state","Code: partial","Code: partial","Code: partial","Offline read behavior needs a single documented contract","Client cache policy only","OMP remains the authority after reconnect" -"H06","T0","Host","Operate pinned unmodified official OMP through the shared adapter","Official OMP exposes RPC command/event seams","Fork bridge remains the released fallback","Gate 0: full","Gate 0: full","Gate 0: full","Official 17.0.6 passes lifecycle, steer/follow-up, approval, cancellation, and no-replay behavior on macOS ARM64 and native Linux x64/ARM64","Plan packaged cutover through the one shared adapter without creating a second adapter","PR #114 used the Lycaon fallback artifact; PR #124 corrects the boundary and retains per-platform evidence" +"H06","T0","Host","Operate pinned unmodified official OMP through the shared adapter","Official OMP exposes RPC command/event seams","Fork bridge remains the released fallback","Gate 0: full","Gate 0: full","Gate 0: full","Official 17.0.6 direct RPC passes on macOS ARM64 and native Linux x64/ARM64; compiled t4-host discovery attach prompt durability and T4-wire projection pass on macOS ARM64","Extend the same exclusive-profile adapter to managed sessions before changing release defaults","PR #124 proves direct Gate 0; this sprint adds packaged-host evidence without a new fork dependency" "H07","T1","Host","Continue from OMP terminal into T4","Terminal command can hand off","Adds /continue-in-t4","Code: full","Code: partial","Code: partial","Mobile landing and failure recovery need proof","Safe CLI handoff plus deep link/session ID","Fork-specific command is appropriate" "H08","T1","Host","Continue from T4 into OMP terminal","Native terminal can resume a session","Same as upstream","Code: partial","Platform","Code: partial","No clear cross-platform handoff action","Explicit CLI handoff with copyable command","Do not emulate the terminal TUI inside the app" -"H09","T0","Host","Official OMP and fallback-bridge release alignment","Current original is 17.0.6","Released fallback bridge is pinned to 17.0.5","Code: partial","Code: partial","Code: partial","The generated Gate 0 snapshot now separates official 17.0.6 from the released Lycaon 17.0.5 fallback; packaged cutover is not proven","Retain exact pins and per-platform evidence, then replace the packaged fallback only after cutover proof","Missing stock seams are explicit: no ready watermark, live session entries, or durable command key" +"H09","T0","Host","Official OMP and fallback-bridge release alignment","Current original is 17.0.6","Released fallback bridge is pinned to 17.0.5","Code: partial","Code: partial","Code: partial","T4 now adapts the missing ready watermark and live entries and has an opt-in compiled local-host proof; managed and shipped defaults still use the fallback","Retain exact pins and both direct and packaged evidence, then replace defaults only after managed proof","Official OMP still lacks a durable command key, so ambiguous dispatch remains outcome-unknown with no automatic replay" "H10","T0","Host","Source vs package vs public release truth","OMP publishes v17.0.6","Fork publishes pinned appserver tag","Main says v0.1.30; public release is v0.1.28","Main says v0.1.30; public release is v0.1.28","Implemented on main; not public release","Release evidence is contradictory","Track on-main, packaged-verified, and publicly-released separately","Public GitHub state checked 2026-07-20" "P01","T0","Projects","Project and session grouping","Sessions retain cwd and project metadata","Authority resolves project roots","Code: full","Code: full","Code: full","No major source gap","Keep projection in T4; roots remain OMP authority","Verify large libraries and moved projects" "P02","T0","Projects","Create project or start session in a project","CLI cwd and new session","Typed session.create and project roots","Code: full","Code: full","Code: full","Needs packaged proof","Existing typed command","Do not create a second project database" diff --git a/docs/adr/019-official-omp-exclusive-profile.md b/docs/adr/019-official-omp-exclusive-profile.md new file mode 100644 index 0000000..ea914ef --- /dev/null +++ b/docs/adr/019-official-omp-exclusive-profile.md @@ -0,0 +1,10 @@ +# ADR-019: Official OMP cutover uses a T4-exclusive profile boundary + +- Status: accepted for opt-in proof; not yet the released default. +- Context: official OMP 17.0.6 exposes the required per-session RPC behavior but not the Lycaon fork's host-wide authority bridge, ready transcript watermark, live durable-entry frames, or cross-process session lock. Making the fork the permanent product center would preserve avoidable coupling; treating arbitrary lockless OMP sessions as writable would be unsafe. +- Decision: T4 may run official OMP through `t4-host` only when the caller supplies an explicit absolute sessions root that is exclusively owned by that T4 profile. T4 owns discovery and lifecycle metadata for that root. Official OMP remains authoritative for each session's JSONL, agent execution, configuration, models, credentials, tools, and events. +- Transcript reconciliation: the shared supervisor streams complete JSONL records, synthesizes the current durable watermark, ignores crash-truncated final records, deduplicates optional fork live frames, and correlates a user entry only when it exactly matches the oldest pending prompt-like command. Ambiguous dispatch remains outcome-unknown and is never automatically replayed. +- Lock boundary: ordinary missing-lock OMP sessions remain view-only. The official profile authority opts into claiming lockless sessions only because the whole configured root is exclusive to T4; it does not describe a missing lock as stale or weaken the final ownership check. +- Compatibility: `t4-host` keeps the Lycaon authority bridge as its default. Official mode requires `--omp-authority official --omp-sessions-root `. The default changes only after packaged desktop and managed-session proofs pass on every required platform. +- Bootstrap boundary: project discovery is currently derived from sessions, so the opt-in path can create more sessions in a known project but cannot expose a project from a completely empty profile. Before this becomes the desktop default, T4 needs a small project registry or an explicit project-add flow that does not invent OMP state. +- Verification: the direct Gate 0 harness proves official RPC lifecycle behavior. The packaged-host harness compiles `t4-host` and proves discovery, attach, promotion, prompt, durable JSONL, and assistant projection over the real T4 Unix-socket protocol. CI retains both evidence artifacts per native platform. diff --git a/package.json b/package.json index 35184fa..a66d3fa 100644 --- a/package.json +++ b/package.json @@ -29,6 +29,7 @@ "package:mac": "node scripts/package-mac-signed.mjs", "stage:omp-runtime:mac": "node scripts/stage-omp-runtime.mjs --platform darwin --arch arm64", "verify:official-omp-lifecycle": "pnpm --filter @t4-code/host-service verify:official-omp-lifecycle", + "verify:official-omp-packaged": "pnpm --filter @t4-code/host-daemon verify:official-omp-packaged", "package:dir": "pnpm prepackage && node scripts/run-electron-builder.mjs --linux --x64 --dir", "inspect:package": "node scripts/inspect-package.mjs", "inspect:dmg": "node scripts/inspect-macos-dmg.mjs", diff --git a/packages/host-daemon/bin/official-omp-packaged-proof.ts b/packages/host-daemon/bin/official-omp-packaged-proof.ts new file mode 100644 index 0000000..78b4661 --- /dev/null +++ b/packages/host-daemon/bin/official-omp-packaged-proof.ts @@ -0,0 +1,204 @@ +#!/usr/bin/env bun + +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import type { ResultFrame, ServerFrame } from "@t4-code/protocol"; +import { OfficialOmpProfileAuthority, profileSocketPath } from "@t4-code/host-service"; +import { startDeterministicModel, verifyRuntime } from "../../host-service/bin/official-omp-gate0.ts"; +import { RawUdsWebSocket } from "../../host-service/test/raw-uds-client.ts"; + +const TIMEOUT_MS = 15_000; + +async function next(client: RawUdsWebSocket): Promise { + return Promise.race([ + client.nextServer(), + Bun.sleep(TIMEOUT_MS).then(() => { + throw new Error("packaged T4 host frame timeout"); + }), + ]); +} + +async function responseFor(client: RawUdsWebSocket, requestId: string): Promise { + for (;;) { + const frame = await next(client); + if (frame.type === "response" && frame.requestId === requestId) return frame; + } +} + +async function waitForSocket(path: string, child: Bun.Subprocess): Promise { + const deadline = Date.now() + TIMEOUT_MS; + while (Date.now() < deadline) { + try { + if ((await stat(path)).isSocket()) return; + } catch {} + const exited = await Promise.race([child.exited.then(code => ({ code })), Bun.sleep(25).then(() => undefined)]); + if (exited) throw new Error(`packaged T4 host exited before its socket was ready (${exited.code})`); + } + throw new Error("packaged T4 host socket timeout"); +} + +async function main(): Promise { + const repoRoot = resolve(import.meta.dirname, "../../.."); + const runtime = await verifyRuntime(repoRoot); + const hostPath = join(repoRoot, "packages", "host-daemon", "dist", "t4-host"); + if (!(await stat(hostPath)).isFile()) throw new Error("build the packaged t4-host before running the proof"); + const root = await mkdtemp(join(tmpdir(), "t4-official-packaged-host-")); + const home = join(root, "home"); + const project = join(root, "project"); + const sessionsRoot = join(root, "sessions"); + const stateRoot = join(root, "state"); + const runtimeRoot = join(root, "run"); + const profile = `gate0-${Bun.randomUUIDv7().slice(-12)}`; + const agentDir = join(home, ".omp", "profiles", profile, "agent"); + const model = startDeterministicModel(); + let child: Bun.Subprocess | undefined; + let client: RawUdsWebSocket | undefined; + try { + await Promise.all([ + mkdir(project, { recursive: true }), + mkdir(agentDir, { recursive: true, mode: 0o700 }), + ]); + await writeFile( + join(agentDir, "models.yml"), + `providers:\n gate0:\n baseUrl: http://127.0.0.1:${model.server.port}/v1\n api: openai-completions\n auth: none\n models:\n - id: deterministic\n name: Gate 0 Deterministic\n reasoning: false\n input: [text]\n contextWindow: 32768\n maxTokens: 4096\n`, + ); + const seed = new OfficialOmpProfileAuthority({ + sessionsRoot, + metadataPath: join(root, "seed-metadata.json"), + }); + await seed.initialize(); + const session = await seed.create(project, "Packaged official OMP"); + await seed.close(); + const socketPath = profileSocketPath(profile, process.platform, home, runtimeRoot); + const environment = { + ...process.env, + HOME: home, + XDG_CONFIG_HOME: join(home, ".config"), + XDG_DATA_HOME: join(home, ".local", "share"), + XDG_STATE_HOME: join(home, ".local", "state"), + XDG_CACHE_HOME: join(home, ".cache"), + XDG_RUNTIME_DIR: runtimeRoot, + PI_NOTIFICATIONS: "off", + OMP_PROFILE: profile, + }; + child = Bun.spawn( + [ + hostPath, + "serve", + "--omp", + runtime.path, + "--omp-authority", + "official", + "--omp-sessions-root", + sessionsRoot, + "--profile", + profile, + "--state-root", + stateRoot, + ], + { + env: environment, + stdout: "pipe", + stderr: "pipe", + }, + ); + const stdout = new Response(child.stdout).text(); + const stderr = new Response(child.stderr).text(); + await waitForSocket(socketPath, child); + client = await RawUdsWebSocket.connect(socketPath); + client.sendJson({ + v: "omp-app/1", + type: "hello", + protocol: { min: "omp-app/1", max: "omp-app/1" }, + client: { name: "official-packaged-proof", version: "1", build: "proof", platform: process.platform }, + requestedFeatures: [], + capabilities: { client: ["sessions.read", "sessions.prompt", "sessions.manage", "catalog.read"] }, + savedCursors: [], + }); + const welcome = await next(client); + if (welcome.type !== "welcome") throw new Error("packaged T4 host did not send Welcome"); + const sessions = await next(client); + if (sessions.type !== "sessions" || !sessions.sessions.some(item => item.sessionId === session.sessionId)) + throw new Error("packaged T4 host did not discover the official OMP session"); + let command = 0; + const send = (requestId: string, name: string, args: Record): void => { + command += 1; + client!.sendJson({ + v: "omp-app/1", + type: "command", + requestId, + commandId: `packaged-${command}`, + hostId: welcome.hostId, + sessionId: session.sessionId, + command: name, + args, + }); + }; + send("attach", "session.attach", {}); + const attached = await responseFor(client, "attach"); + if (!attached.ok) throw new Error(`packaged session attach failed: ${attached.error.message}`); + let stateReady = false; + let stateFailure = "unknown"; + for (let attempt = 0; attempt < 40 && !stateReady; attempt += 1) { + const requestId = `state-${attempt}`; + send(requestId, "session.state.get", {}); + const state = await responseFor(client, requestId); + stateReady = state.ok; + if (!state.ok) stateFailure = `${state.error.code}: ${state.error.message}`; + if (!stateReady) await Bun.sleep(100); + } + if (!stateReady) throw new Error(`packaged session did not become writable (${stateFailure})`); + send("prompt", "session.prompt", { message: "Packaged host prompt" }); + const prompted = await responseFor(client, "prompt"); + if (!prompted.ok) throw new Error(`packaged session prompt failed: ${prompted.error.message}`); + let assistantProjected = false; + const deadline = Date.now() + TIMEOUT_MS; + while (Date.now() < deadline && !assistantProjected) { + const frame = await next(client); + assistantProjected = + frame.type === "entry" && + frame.sessionId === session.sessionId && + frame.entry.kind === "message" && + frame.entry.data.role === "assistant" && + frame.entry.data.text === "Gate 0 response 1"; + } + if (!assistantProjected) throw new Error("official OMP assistant entry did not reach the T4 wire"); + const transcript = await readFile(session.path, "utf8"); + if (!transcript.includes("Packaged host prompt") || !transcript.includes("Gate 0 response 1")) + throw new Error("packaged host turn was not durable in official OMP JSONL"); + const result = { + schemaVersion: 1, + runtime: { + version: runtime.version, + tag: runtime.matrix.officialRuntime.sourceTag, + commit: runtime.matrix.officialRuntime.sourceCommit, + sha256: runtime.manifest.sha256, + }, + platform: { os: process.platform, arch: process.arch }, + packagedHost: { binary: "t4-host", authority: "official", exclusiveSessionsRoot: true }, + scenarios: { discovery: true, attach: true, prompt: true, durableJsonl: true, t4WireProjection: true }, + passed: true, + }; + const evidenceRoot = join(repoRoot, "artifacts", "official-omp-packaged-host"); + await mkdir(evidenceRoot, { recursive: true }); + await writeFile(join(evidenceRoot, `${process.platform}-${process.arch}.json`), `${JSON.stringify(result, null, 2)}\n`); + console.log(JSON.stringify(result, null, 2)); + await client.close(); + client = undefined; + child.kill("SIGTERM"); + if ((await child.exited) !== 0) throw new Error(`packaged T4 host failed: ${(await stderr).trim().slice(-4_096)}`); + child = undefined; + await stdout; + } finally { + client?.destroy(); + if (child) { + child.kill("SIGKILL"); + await child.exited.catch(() => undefined); + } + await model.server.stop(true); + await rm(root, { recursive: true, force: true }); + } +} + +await main(); diff --git a/packages/host-daemon/package.json b/packages/host-daemon/package.json index d92b26e..7e66843 100644 --- a/packages/host-daemon/package.json +++ b/packages/host-daemon/package.json @@ -11,11 +11,12 @@ "build:binary": "bun build --compile src/cli.ts --outfile dist/t4-host", "check": "bun run typecheck", "test": "bun test test", + "verify:official-omp-packaged": "node ../../scripts/stage-omp-runtime.mjs --runtime official && bun run build:binary && bun run bin/official-omp-packaged-proof.ts", "typecheck": "tsgo -p tsconfig.json --noEmit" }, "dependencies": { "@t4-code/host-service": "workspace:*", - "@t4-code/host-wire": "workspace:*" + "@t4-code/protocol": "workspace:*" }, "devDependencies": { "@types/bun": "1.3.5", diff --git a/packages/host-daemon/src/cli.ts b/packages/host-daemon/src/cli.ts index 216e64b..b6700eb 100644 --- a/packages/host-daemon/src/cli.ts +++ b/packages/host-daemon/src/cli.ts @@ -6,20 +6,55 @@ import { isAbsolute, join, resolve } from "node:path"; import { createAppserver, createRemoteAppserver, + OfficialOmpProfileAuthority, OmpAuthorityBridgeClient, profileSocketPath, ProjectFileSearchAuthority, TranscriptSearchIndex, type AppserverHandle, type AppserverOptions, + type DesktopOperationsAuthority, + type SessionAuthority, + type SessionDiscovery, } from "@t4-code/host-service"; +import { COMMAND_DESCRIPTORS, type ProjectId, type SessionId } from "@t4-code/protocol"; export const T4_HOST_VERSION = "0.1.30"; +export const OFFICIAL_OMP_VERSION = "17.0.6"; +export const OFFICIAL_OMP_BUILD = "89d6a8f6d14286f32f09ec9c8aa8af7b3451d2d6"; const PROFILE = /^[a-z0-9][a-z0-9._-]{0,63}$/u; const ORIGIN_LIMIT = 32; +const VERSION_OUTPUT_BYTES = 4 * 1024; +const VERSION_TIMEOUT_MS = 5_000; +const OFFICIAL_CATALOG_COMMANDS = Object.freeze([ + "session.create", + "session.rename", + "session.archive", + "session.restore", + "session.delete", + "session.model.set", + "session.thinking.set", + "session.cancel", + "session.close", +]); + +function officialCatalogItems(): Record[] { + const commands = process.platform === "darwin" + ? ["project.reveal", ...OFFICIAL_CATALOG_COMMANDS] + : OFFICIAL_CATALOG_COMMANDS; + return commands.map(name => ({ + id: `cmd-${name.replaceAll(".", "-")}`, + kind: "command", + name, + capabilities: [COMMAND_DESCRIPTORS[name]!.capability], + supported: true, + })); +} export interface HostDaemonConfig { readonly ompExecutable: string; + readonly authorityMode?: "bridge" | "official"; + readonly ompSessionsRoot?: string; readonly profileId: string; readonly stateRoot: string; readonly remote?: { @@ -36,6 +71,7 @@ export interface HostDaemonPaths { readonly hostIdPath: string; readonly attentionOutcomePath: string; readonly transcriptSearchPath: string; + readonly officialMetadataPath: string; readonly remoteStateRoot: string; readonly socketPath: string; } @@ -65,6 +101,8 @@ function boundedOrigin(input: string): string { export function parseHostDaemonArgs(argv: readonly string[], home = homedir()): HostDaemonConfig { if (argv[0] !== "serve") throw new Error("t4-host requires the serve action"); let ompExecutable: string | undefined; + let authorityMode: "bridge" | "official" = "bridge"; + let ompSessionsRoot: string | undefined; let profileId = "default"; let stateRoot = join(home, ".t4-code", "host"); let remoteMode: "direct" | "serve" | undefined; @@ -75,6 +113,12 @@ export function parseHostDaemonArgs(argv: readonly string[], home = homedir()): for (let index = 1; index < argv.length; index += 1) { const flag = argv[index]!; if (flag === "--omp") ompExecutable = value(argv, index++, flag); + else if (flag === "--omp-authority") { + const mode = value(argv, index++, flag); + if (mode !== "bridge" && mode !== "official") + throw new Error("--omp-authority must be bridge or official"); + authorityMode = mode; + } else if (flag === "--omp-sessions-root") ompSessionsRoot = value(argv, index++, flag); else if (flag === "--profile") profileId = value(argv, index++, flag); else if (flag === "--state-root") stateRoot = value(argv, index++, flag); else if (flag === "--remote-mode") { @@ -97,6 +141,10 @@ export function parseHostDaemonArgs(argv: readonly string[], home = homedir()): throw new Error("--omp must name an absolute executable path"); if (!PROFILE.test(profileId)) throw new Error("--profile is invalid"); if (!isAbsolute(stateRoot)) throw new Error("--state-root must be absolute"); + if (authorityMode === "official" && (!ompSessionsRoot || !isAbsolute(ompSessionsRoot))) + throw new Error("official OMP authority requires an absolute --omp-sessions-root"); + if (authorityMode === "bridge" && ompSessionsRoot) + throw new Error("--omp-sessions-root requires official OMP authority"); if (!remoteMode && (remoteAddress || origins.length || trustedServeProxy || remotePort !== 8787)) throw new Error("remote flags require --remote-mode"); if (remoteMode && !remoteAddress) throw new Error("remote mode requires --remote-address"); @@ -108,6 +156,8 @@ export function parseHostDaemonArgs(argv: readonly string[], home = homedir()): throw new Error("trusted Serve proxy is invalid in direct mode"); return { ompExecutable: resolve(ompExecutable), + authorityMode, + ...(ompSessionsRoot ? { ompSessionsRoot: resolve(ompSessionsRoot) } : {}), profileId, stateRoot: resolve(stateRoot), ...(remoteMode @@ -137,6 +187,7 @@ export function hostDaemonPaths( hostIdPath: join(profileStateRoot, "host-id"), attentionOutcomePath: join(profileStateRoot, "attention-outcomes.json"), transcriptSearchPath: join(profileStateRoot, "transcript-search.sqlite"), + officialMetadataPath: join(profileStateRoot, "official-omp-sessions.json"), remoteStateRoot: join(profileStateRoot, "remote"), socketPath: profileSocketPath(config.profileId), }; @@ -144,36 +195,141 @@ export function hostDaemonPaths( export interface HostDaemonDependencies { readonly createBridge?: (config: HostDaemonConfig) => OmpAuthorityBridgeClient; + readonly createOfficialAuthority?: ( + config: HostDaemonConfig, + paths: HostDaemonPaths, + ) => OfficialOmpProfileAuthority; readonly createTranscriptSearch?: (path: string) => TranscriptSearchIndex; readonly createLocal?: (options: AppserverOptions) => AppserverHandle; readonly createRemote?: typeof createRemoteAppserver; + readonly verifyOfficialRuntime?: (executable: string) => Promise>; readonly onSignal?: (signal: "SIGINT" | "SIGTERM", listener: () => void) => void; readonly removeSignal?: (signal: "SIGINT" | "SIGTERM", listener: () => void) => void; } +async function boundedProcessOutput(stream: ReadableStream, maxBytes: number): Promise { + const reader = stream.getReader(); + const chunks: Uint8Array[] = []; + let size = 0; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + size += value.byteLength; + if (size > maxBytes) throw new Error("official OMP version output exceeds 4 KiB"); + chunks.push(value); + } + } finally { + reader.releaseLock(); + } + const output = new Uint8Array(size); + let offset = 0; + for (const chunk of chunks) { + output.set(chunk, offset); + offset += chunk.byteLength; + } + return new TextDecoder("utf-8", { fatal: true }).decode(output); +} + +export async function verifyOfficialRuntime( + executable: string, +): Promise> { + const child = Bun.spawn([executable, "--version"], { + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + env: {}, + }); + const timer = setTimeout(() => child.kill(), VERSION_TIMEOUT_MS); + try { + const [stdout, stderr, exitCode] = await Promise.all([ + boundedProcessOutput(child.stdout, VERSION_OUTPUT_BYTES), + boundedProcessOutput(child.stderr, VERSION_OUTPUT_BYTES), + child.exited, + ]); + if (exitCode !== 0) throw new Error(`official OMP version probe failed (${exitCode}): ${stderr.trim()}`); + if (stdout.trim() !== `omp/${OFFICIAL_OMP_VERSION}`) + throw new Error(`official OMP runtime must report omp/${OFFICIAL_OMP_VERSION}`); + return { ompVersion: OFFICIAL_OMP_VERSION, ompBuild: OFFICIAL_OMP_BUILD }; + } finally { + clearTimeout(timer); + if (child.exitCode === null) child.kill(); + } +} + export async function runHostDaemon( config: HostDaemonConfig, dependencies: HostDaemonDependencies = {}, ): Promise { const paths = hostDaemonPaths(config); await mkdir(paths.profileStateRoot, { recursive: true, mode: 0o700 }); - const bridge = - dependencies.createBridge?.(config) ?? - new OmpAuthorityBridgeClient({ - executable: config.ompExecutable, - environment: { OMP_PROFILE: config.profileId }, - }); - await bridge.start(); + let bridge: OmpAuthorityBridgeClient | undefined; + let officialAuthority: OfficialOmpProfileAuthority | undefined; + let sessionAuthority: SessionAuthority; + let discovery: SessionDiscovery; + let operationsAuthority: DesktopOperationsAuthority = {}; + let usageAuthority: AppserverOptions["usageAuthority"]; + let transcriptImageRoot: string | undefined; + let identity: Pick = {}; + let projectRootForProject: (projectId: ProjectId) => Promise | string; + let projectRootForSession: (sessionId: SessionId) => Promise; + let lockCheck: NonNullable; + let lockStatus: NonNullable; + if (config.authorityMode === "official") { + identity = await (dependencies.verifyOfficialRuntime ?? verifyOfficialRuntime)(config.ompExecutable); + const official = + dependencies.createOfficialAuthority?.(config, paths) ?? + new OfficialOmpProfileAuthority({ + sessionsRoot: config.ompSessionsRoot!, + metadataPath: paths.officialMetadataPath, + }); + await official.initialize(); + officialAuthority = official; + sessionAuthority = official; + discovery = official; + operationsAuthority = { + catalogGet: async () => ({ + revision: `official-omp-${OFFICIAL_OMP_VERSION}`, + items: officialCatalogItems(), + }), + }; + projectRootForProject = projectId => official.projectRootForProject(projectId); + projectRootForSession = sessionId => official.projectRootForSession(sessionId); + lockCheck = session => official.lockCheck(session); + lockStatus = () => official.lockStatus(); + } else { + bridge = + dependencies.createBridge?.(config) ?? + new OmpAuthorityBridgeClient({ + executable: config.ompExecutable, + environment: { OMP_PROFILE: config.profileId }, + }); + try { + await bridge.start(); + const authorities = bridge.createAuthorities(); + const hostInfo = await authorities.hostInfo(); + sessionAuthority = authorities.sessionAuthority; + discovery = authorities.discovery; + operationsAuthority = authorities.operationsAuthority; + usageAuthority = authorities.usageAuthority; + transcriptImageRoot = hostInfo.transcriptImageRoot; + identity = bridge.identity; + projectRootForProject = authorities.projectRootForProject; + projectRootForSession = authorities.projectRootForSession; + lockCheck = authorities.lockCheck; + lockStatus = authorities.lockStatus; + } catch (error) { + await bridge.stop(); + throw error; + } + } try { - const authorities = bridge.createAuthorities(); - const hostInfo = await authorities.hostInfo(); const transcriptSearchAuthority = dependencies.createTranscriptSearch?.(paths.transcriptSearchPath) ?? new TranscriptSearchIndex(paths.transcriptSearchPath); const projectFileSearchAuthority = new ProjectFileSearchAuthority( - authorities.projectRootForSession, + projectRootForSession, ); - const identity = bridge.identity; const options: AppserverOptions = { ...identity, appserverVersion: T4_HOST_VERSION, @@ -181,19 +337,22 @@ export async function runHostDaemon( socketPath: paths.socketPath, hostIdPath: paths.hostIdPath, attentionOutcomePath: paths.attentionOutcomePath, - sessionAuthority: authorities.sessionAuthority, - discovery: authorities.discovery, + sessionAuthority, + discovery, operationsAuthority: { - ...authorities.operationsAuthority, + ...operationsAuthority, ...projectFileSearchAuthority.operations(), }, - usageAuthority: authorities.usageAuthority, + ...(usageAuthority ? { usageAuthority } : {}), transcriptSearchAuthority, - projectRootForProject: authorities.projectRootForProject, - lockCheck: authorities.lockCheck, - lockStatus: authorities.lockStatus, - transcriptImageRoot: hostInfo.transcriptImageRoot, + projectRootForProject, + lockCheck, + lockStatus, + ...(config.authorityMode === "official" ? { claimLocklessSessions: true } : {}), + ...(transcriptImageRoot ? { transcriptImageRoot } : {}), rpcChildInvocation: { executable: config.ompExecutable, prefixArgv: [] }, + rpcChildEnvironment: { OMP_PROFILE: config.profileId }, + ...(config.authorityMode === "official" ? { rpcDialect: "official-17.0.6" as const } : {}), ...(process.platform === "darwin" ? { projectRevealer: async (root: string): Promise => { @@ -246,7 +405,8 @@ export async function runHostDaemon( if (!stopping) await appserver.stop().catch(() => undefined); } } finally { - await bridge.stop(); + await bridge?.stop(); + await officialAuthority?.close(); } } diff --git a/packages/host-daemon/test/cli.test.ts b/packages/host-daemon/test/cli.test.ts index ad18a23..2597ba8 100644 --- a/packages/host-daemon/test/cli.test.ts +++ b/packages/host-daemon/test/cli.test.ts @@ -1,5 +1,15 @@ import { describe, expect, test } from "bun:test"; -import { hostDaemonPaths, parseHostDaemonArgs, runHostDaemon } from "../src/cli.ts"; +import { chmod, mkdtemp, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + hostDaemonPaths, + OFFICIAL_OMP_BUILD, + OFFICIAL_OMP_VERSION, + parseHostDaemonArgs, + runHostDaemon, + verifyOfficialRuntime, +} from "../src/cli.ts"; describe("T4 host daemon CLI", () => { test("parses a local direct-replacement service without ambient executable lookup", () => { @@ -9,6 +19,7 @@ describe("T4 host daemon CLI", () => { ); expect(config).toEqual({ ompExecutable: "/opt/t4/runtime/omp", + authorityMode: "bridge", profileId: "default", stateRoot: "/home/test/.t4-code/host", }); @@ -49,6 +60,32 @@ describe("T4 host daemon CLI", () => { "/home/test", ), ).toThrow("HTTP origin"); + expect(() => + parseHostDaemonArgs( + ["serve", "--omp", "/opt/omp", "--omp-authority", "official"], + "/home/test", + ), + ).toThrow("--omp-sessions-root"); + expect( + parseHostDaemonArgs( + [ + "serve", + "--omp", + "/opt/omp", + "--omp-authority", + "official", + "--omp-sessions-root", + "/home/test/.omp/t4/sessions", + "--profile", + "t4", + ], + "/home/test", + ), + ).toMatchObject({ + authorityMode: "official", + ompSessionsRoot: "/home/test/.omp/t4/sessions", + profileId: "t4", + }); }); test("stops the OMP bridge when authority startup fails", async () => { @@ -97,4 +134,76 @@ describe("T4 host daemon CLI", () => { expect(searchCloses).toBe(1); expect(bridgeStops).toBe(1); }); + + test("pins and reports the exact official OMP runtime before exposing official authority", async () => { + let authorityCloses = 0; + let captured: Record | undefined; + const authority = { + initialize: async () => {}, + close: async () => { authorityCloses += 1; }, + projectRootForProject: async () => "/tmp", + projectRootForSession: async () => "/tmp", + lockCheck: async () => {}, + lockStatus: () => "missing", + list: async () => [], + }; + await expect( + runHostDaemon( + { + ompExecutable: "/opt/omp", + authorityMode: "official", + ompSessionsRoot: "/tmp/t4-official-sessions", + profileId: "t4", + stateRoot: "/tmp/t4-official-state", + }, + { + verifyOfficialRuntime: async () => ({ + ompVersion: OFFICIAL_OMP_VERSION, + ompBuild: OFFICIAL_OMP_BUILD, + }), + createOfficialAuthority: () => authority as never, + createTranscriptSearch: () => ({ close: async () => {} }) as never, + createLocal: options => { + captured = options as unknown as Record; + throw new Error("captured official options"); + }, + }, + ), + ).rejects.toThrow("captured official options"); + expect(captured).toMatchObject({ + ompVersion: OFFICIAL_OMP_VERSION, + ompBuild: OFFICIAL_OMP_BUILD, + rpcDialect: "official-17.0.6", + claimLocklessSessions: true, + }); + const operations = captured?.operationsAuthority as { + catalogGet?: () => Promise>; + }; + expect(await operations.catalogGet?.()).toMatchObject({ + revision: `official-omp-${OFFICIAL_OMP_VERSION}`, + }); + const catalog = await operations.catalogGet?.(); + if (!catalog) throw new Error("official catalog missing"); + const commandNames = (catalog.items as Array<{ name: string }>).map(item => item.name); + expect(commandNames).toContain("session.model.set"); + expect(commandNames).not.toContain("session.fast.set"); + expect(commandNames).not.toContain("session.retry"); + expect(authorityCloses).toBe(1); + }); + + test("official runtime probe fails closed on version drift", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-version-")); + const exact = join(root, "exact-omp"); + const drifted = join(root, "drifted-omp"); + await Promise.all([ + writeFile(exact, `#!/bin/sh\nprintf 'omp/${OFFICIAL_OMP_VERSION}\\n'\n`), + writeFile(drifted, "#!/bin/sh\nprintf 'omp/17.0.7\\n'\n"), + ]); + await Promise.all([chmod(exact, 0o700), chmod(drifted, 0o700)]); + expect(await verifyOfficialRuntime(exact)).toEqual({ + ompVersion: OFFICIAL_OMP_VERSION, + ompBuild: OFFICIAL_OMP_BUILD, + }); + await expect(verifyOfficialRuntime(drifted)).rejects.toThrow(`omp/${OFFICIAL_OMP_VERSION}`); + }); }); diff --git a/packages/host-service/bin/official-omp-gate0.ts b/packages/host-service/bin/official-omp-gate0.ts index 8bb1766..1c5643d 100755 --- a/packages/host-service/bin/official-omp-gate0.ts +++ b/packages/host-service/bin/official-omp-gate0.ts @@ -51,7 +51,7 @@ interface TranscriptWatermark { readonly entryCount: number; } -interface DeterministicModel { +export interface DeterministicModel { readonly server: Bun.Server; readonly requests: string[][]; readonly gateNextRequest: () => ModelGate; @@ -223,7 +223,7 @@ function requestMessages(body: JsonMap): string[] { }); } -function startDeterministicModel(): DeterministicModel { +export function startDeterministicModel(): DeterministicModel { const requests: string[][] = []; let nextGate: | { @@ -555,7 +555,7 @@ async function runApprovalScenario(input: { } } -async function verifyRuntime(repoRoot: string): Promise { +export async function verifyRuntime(repoRoot: string): Promise { const matrix = decodeRuntimeMatrix(await readJson(join(repoRoot, "compat", "omp-app-matrix.json"))); const manifestPath = join(repoRoot, ".artifacts", "omp-runtime-official", "manifest.json"); const manifest = decodeRuntimeManifest(await readJson(manifestPath)); @@ -728,4 +728,4 @@ async function main(): Promise { } } -await main(); +if (import.meta.main) await main(); diff --git a/packages/host-service/src/index.ts b/packages/host-service/src/index.ts index 02ce50d..5330727 100644 --- a/packages/host-service/src/index.ts +++ b/packages/host-service/src/index.ts @@ -7,6 +7,7 @@ export * from "./image-upload-store.ts"; export * from "./omp-authority-bridge-contract.ts"; export * from "./omp-authority-bridge-client.ts"; export * from "./official-omp-capabilities.ts"; +export * from "./official-omp-profile-authority.ts"; export * from "./project-file-search.ts"; export * from "./operations/index.ts"; export * from "./projection.ts"; diff --git a/packages/host-service/src/official-omp-profile-authority.ts b/packages/host-service/src/official-omp-profile-authority.ts new file mode 100644 index 0000000..8b1e792 --- /dev/null +++ b/packages/host-service/src/official-omp-profile-authority.ts @@ -0,0 +1,472 @@ +import { chmod, link, lstat, mkdir, open, readFile, readdir, realpath, rename, rm, stat, unlink } from "node:fs/promises"; +import { dirname, isAbsolute, join, relative, resolve } from "node:path"; +import { + hostId, + type ProjectId, + type SessionId, + sessionId, + type TranscriptPageArguments, + type TranscriptPageResult, +} from "@t4-code/host-wire"; +import { FileSessionDiscovery } from "./discovery.ts"; +import type { + SessionAuthority, + SessionAuthoritySession, + SessionDiscovery, + SessionRecord, +} from "./types.ts"; + +const TITLE_SLOT_BYTES = 256; +const METADATA_BYTES = 1024 * 1024; +const OWNER_FILE = ".t4-exclusive-owner.lock"; + +interface OwnerRecord { + readonly version: 1; + readonly pid: number; + readonly ownerId: string; +} + +interface FileIdentity { + readonly device: number; + readonly inode: number; +} + +interface OfficialProfileMetadata { + readonly version: 1; + readonly archived: Readonly>; +} + +export interface OfficialOmpProfileAuthorityOptions { + readonly sessionsRoot: string; + readonly metadataPath: string; +} + +function titleSlot(title: string, updatedAt: string): string { + const encoder = new TextEncoder(); + const codePoints = [...title]; + const line = (value: string, pad: string): string => + `${JSON.stringify({ type: "title", v: 1, title: value, source: "user", updatedAt, pad })}\n`; + let low = 0; + let high = codePoints.length; + let bounded = ""; + while (low <= high) { + const middle = (low + high) >>> 1; + const candidate = codePoints.slice(0, middle).join(""); + if (encoder.encode(line(candidate, "")).byteLength <= TITLE_SLOT_BYTES) { + bounded = candidate; + low = middle + 1; + } else high = middle - 1; + } + const unpadded = line(bounded, ""); + const pad = " ".repeat(TITLE_SLOT_BYTES - encoder.encode(unpadded).byteLength); + const result = line(bounded, pad); + if (encoder.encode(result).byteLength !== TITLE_SLOT_BYTES) throw new Error("official OMP title slot is invalid"); + return result; +} + +function decodeMetadata(value: unknown): OfficialProfileMetadata { + if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error("official OMP metadata is invalid"); + const root = value as Record; + if (root.version !== 1 || !root.archived || typeof root.archived !== "object" || Array.isArray(root.archived)) + throw new Error("official OMP metadata is invalid"); + const archived: Record = {}; + for (const [id, timestamp] of Object.entries(root.archived)) { + if (id.length === 0 || id.length > 256 || typeof timestamp !== "string" || !Number.isFinite(Date.parse(timestamp))) + throw new Error("official OMP metadata is invalid"); + archived[id] = timestamp; + } + return { version: 1, archived }; +} + +function decodeOwnerRecord(value: unknown): OwnerRecord { + if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error("official OMP owner lease is invalid"); + const record = value as Record; + if ( + record.version !== 1 || + typeof record.pid !== "number" || + !Number.isSafeInteger(record.pid) || + record.pid <= 0 || + typeof record.ownerId !== "string" || + !/^[0-9a-f-]{36}$/u.test(record.ownerId) + ) + throw new Error("official OMP owner lease is invalid"); + return { version: 1, pid: record.pid, ownerId: record.ownerId }; +} + +function processIsAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + +async function syncDirectory(path: string): Promise { + const handle = await open(path, "r"); + try { + await handle.sync(); + } finally { + await handle.close(); + } +} + +/** + * T4-owned host management for an isolated official-OMP profile. The caller + * must give this authority an exclusive sessions root because stock OMP has no + * cross-process writer lock. OMP remains the per-session runtime and JSONL + * authority; this class supplies only host-wide discovery and lifecycle seams. + */ +export class OfficialOmpProfileAuthority implements SessionAuthority, SessionDiscovery { + readonly #sessionsRoot: string; + readonly #metadataPath: string; + readonly #discovery: FileSessionDiscovery; + readonly #archived = new Map(); + readonly #owner: OwnerRecord = { version: 1, pid: process.pid, ownerId: Bun.randomUUIDv7() }; + #canonicalRoot?: string; + #ownerIdentity?: FileIdentity; + #metadataMutation: Promise = Promise.resolve(); + #initialized = false; + + constructor(options: OfficialOmpProfileAuthorityOptions) { + if (!isAbsolute(options.sessionsRoot) || !isAbsolute(options.metadataPath)) + throw new Error("official OMP authority paths must be absolute"); + this.#sessionsRoot = resolve(options.sessionsRoot); + this.#metadataPath = resolve(options.metadataPath); + this.#discovery = new FileSessionDiscovery(this.#sessionsRoot, undefined, hostId("official-omp"), true); + } + + async initialize(): Promise { + if (this.#initialized) return; + await Promise.all([ + mkdir(this.#sessionsRoot, { recursive: true, mode: 0o700 }), + mkdir(dirname(this.#metadataPath), { recursive: true, mode: 0o700 }), + ]); + const rootInfo = await lstat(this.#sessionsRoot); + const metadataRootInfo = await lstat(dirname(this.#metadataPath)); + const uid = process.getuid?.(); + if ( + !rootInfo.isDirectory() || + rootInfo.isSymbolicLink() || + !metadataRootInfo.isDirectory() || + metadataRootInfo.isSymbolicLink() || + (uid !== undefined && (rootInfo.uid !== uid || metadataRootInfo.uid !== uid)) + ) + throw new Error("official OMP authority root is unsafe"); + await Promise.all([chmod(this.#sessionsRoot, 0o700), chmod(dirname(this.#metadataPath), 0o700)]); + this.#canonicalRoot = await realpath(this.#sessionsRoot); + await this.#acquireLease(); + try { + try { + const info = await lstat(this.#metadataPath); + if (info.isSymbolicLink() || !info.isFile() || info.size > METADATA_BYTES) + throw new Error("official OMP metadata is invalid"); + const metadata = decodeMetadata(JSON.parse(await readFile(this.#metadataPath, "utf8"))); + for (const [id, timestamp] of Object.entries(metadata.archived)) this.#archived.set(id, timestamp); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + this.#initialized = true; + } catch (error) { + await this.close(); + throw error; + } + } + + async close(): Promise { + if (!this.#ownerIdentity || !this.#canonicalRoot) return; + await this.#metadataMutation.catch(() => undefined); + const ownerPath = join(this.#canonicalRoot, OWNER_FILE); + try { + const info = await lstat(ownerPath); + const current = decodeOwnerRecord(JSON.parse(await readFile(ownerPath, "utf8"))); + if ( + info.dev === this.#ownerIdentity.device && + info.ino === this.#ownerIdentity.inode && + current.pid === this.#owner.pid && + current.ownerId === this.#owner.ownerId + ) { + await unlink(ownerPath); + await syncDirectory(this.#canonicalRoot); + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } finally { + this.#ownerIdentity = undefined; + this.#initialized = false; + } + } + + async list(): Promise { + await this.#assertLease(); + await this.#assertDiscoveryTree(); + const records = await this.#discovery.list(); + await Promise.all(records.map(record => this.#assertOwnedSession(record))); + return records.map(record => { + const archivedAt = this.#archived.get(record.sessionId); + return archivedAt ? { ...record, archivedAt } : record; + }); + } + + async load(session: SessionRecord): Promise { + await this.#assertLease(); + await this.#assertOwnedSession(session); + const loaded = await this.#discovery.load(session); + await this.#assertOwnedSession(loaded); + const archivedAt = this.#archived.get(loaded.sessionId); + return archivedAt ? { ...loaded, archivedAt } : loaded; + } + + async page(session: SessionRecord, args: TranscriptPageArguments): Promise { + await this.#assertLease(); + await this.#assertOwnedSession(session); + if (!this.#discovery.page) throw new Error("official OMP transcript paging is unavailable"); + return this.#discovery.page(session, args); + } + + async create(cwd: string, title = "Session"): Promise { + await this.#assertLease(); + const canonicalCwd = await realpath(cwd); + if (!(await stat(canonicalCwd)).isDirectory()) throw new Error("official OMP session cwd is unavailable"); + const id = Bun.randomUUIDv7(); + const timestamp = new Date().toISOString(); + const configuredDirectory = join(this.#sessionsRoot, "-t4"); + await mkdir(configuredDirectory, { recursive: true, mode: 0o700 }); + const directory = await this.#assertOwnedDirectory(configuredDirectory); + const path = join(directory, `session-${id}.jsonl`); + const body = `${titleSlot(title, timestamp)}${JSON.stringify({ + type: "session", + version: 3, + id, + timestamp, + cwd: canonicalCwd, + })}\n`; + const handle = await open(path, "wx", 0o600); + try { + await handle.writeFile(body, "utf8"); + await handle.sync(); + } finally { + await handle.close(); + } + return { sessionId: sessionId(id), path, cwd: canonicalCwd, title, entries: [] }; + } + + async archive(session: SessionRecord, archivedAt: string): Promise { + await this.#assertLease(); + await this.#assertOwnedSession(session); + await this.#mutateMetadata(async () => { + const previous = this.#archived.get(session.sessionId); + this.#archived.set(session.sessionId, archivedAt); + try { + await this.#persist(); + } catch (error) { + if (previous === undefined) this.#archived.delete(session.sessionId); + else this.#archived.set(session.sessionId, previous); + throw error; + } + }); + } + + async restore(session: SessionRecord): Promise { + await this.#assertLease(); + await this.#assertOwnedSession(session); + await this.#mutateMetadata(async () => { + const previous = this.#archived.get(session.sessionId); + this.#archived.delete(session.sessionId); + try { + await this.#persist(); + } catch (error) { + if (previous !== undefined) this.#archived.set(session.sessionId, previous); + throw error; + } + }); + } + + async delete(session: SessionRecord): Promise { + await this.#assertLease(); + const path = await this.#assertOwnedSession(session); + const artifacts = path.slice(0, -".jsonl".length); + let artifactsExist = false; + try { + const info = await lstat(artifacts); + const uid = process.getuid?.(); + if (info.isSymbolicLink() || !info.isDirectory() || (uid !== undefined && info.uid !== uid)) + throw new Error("official OMP artifact root is unsafe"); + const canonicalArtifacts = await realpath(artifacts); + const child = relative(this.#canonicalRoot!, canonicalArtifacts); + if (child === "" || child.startsWith("..") || isAbsolute(child)) + throw new Error("official OMP artifact root is outside the exclusive root"); + artifactsExist = true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + const trash = join(this.#canonicalRoot!, ".t4-trash"); + await mkdir(trash, { mode: 0o700 }); + await this.#assertOwnedDirectory(trash); + const nonce = Bun.randomUUIDv7(); + const trashedPath = join(trash, `${nonce}.jsonl`); + const trashedArtifacts = join(trash, nonce); + await rename(path, trashedPath); + try { + if (artifactsExist) await rename(artifacts, trashedArtifacts); + } catch (error) { + await rename(trashedPath, path).catch(() => undefined); + throw error; + } + await syncDirectory(dirname(path)); + await syncDirectory(trash); + await this.#mutateMetadata(async () => { + this.#archived.delete(session.sessionId); + await this.#persist().catch(() => undefined); + }); + await Promise.all([ + rm(trashedPath, { force: true }), + artifactsExist ? rm(trashedArtifacts, { recursive: true, force: true }) : Promise.resolve(), + ]); + } + + async projectRootForProject(project: ProjectId): Promise { + const roots = new Set((await this.list()).filter(record => record.projectId === project).map(record => record.cwd)); + if (roots.size !== 1) throw new Error("official OMP project root is unavailable"); + return [...roots][0]!; + } + + async projectRootForSession(id: SessionId): Promise { + const session = (await this.list()).find(record => record.sessionId === id); + if (!session) throw new Error("official OMP session root is unavailable"); + return session.cwd; + } + + lockStatus(): "missing" { + return "missing"; + } + + async lockCheck(session: SessionRecord): Promise { + await this.#assertLease(); + await this.#assertOwnedSession(session); + } + + #assertInitialized(): void { + if (!this.#initialized || !this.#canonicalRoot) throw new Error("official OMP authority is not initialized"); + } + + async #assertOwnedSession(session: SessionRecord): Promise { + this.#assertInitialized(); + if (!session.path.endsWith(".jsonl")) throw new Error("official OMP session path is invalid"); + const pathInfo = await lstat(session.path); + if (pathInfo.isSymbolicLink() || !pathInfo.isFile()) throw new Error("official OMP session path is invalid"); + const canonical = await realpath(session.path); + const child = relative(this.#canonicalRoot!, canonical); + if (child === "" || child.startsWith("..") || isAbsolute(child)) + throw new Error("official OMP session is outside the exclusive root"); + return canonical; + } + + async #assertDiscoveryTree(): Promise { + for (const entry of await readdir(this.#canonicalRoot!, { withFileTypes: true })) { + const path = join(this.#canonicalRoot!, entry.name); + const info = await lstat(path); + if (info.isSymbolicLink()) throw new Error("official OMP discovery tree contains a symlink"); + if (!info.isDirectory() || entry.name === ".t4-trash") continue; + if (!entry.name.startsWith("-")) continue; + await this.#assertOwnedDirectory(path); + for (const child of await readdir(path, { withFileTypes: true })) { + const childInfo = await lstat(join(path, child.name)); + if (childInfo.isSymbolicLink()) throw new Error("official OMP discovery tree contains a symlink"); + } + } + } + + async #assertLease(): Promise { + this.#assertInitialized(); + if (!this.#ownerIdentity) throw new Error("official OMP exclusive owner lease is unavailable"); + const ownerPath = join(this.#canonicalRoot!, OWNER_FILE); + const info = await lstat(ownerPath); + const record = decodeOwnerRecord(JSON.parse(await readFile(ownerPath, "utf8"))); + if ( + info.isSymbolicLink() || + !info.isFile() || + info.dev !== this.#ownerIdentity.device || + info.ino !== this.#ownerIdentity.inode || + record.pid !== this.#owner.pid || + record.ownerId !== this.#owner.ownerId + ) + throw new Error("official OMP exclusive owner lease changed"); + } + + async #acquireLease(): Promise { + const ownerPath = join(this.#canonicalRoot!, OWNER_FILE); + for (let attempt = 0; attempt < 2; attempt += 1) { + const temporary = join(this.#canonicalRoot!, `.t4-owner-${this.#owner.ownerId}.tmp`); + const handle = await open(temporary, "wx", 0o600); + try { + await handle.writeFile(`${JSON.stringify(this.#owner)}\n`, "utf8"); + await handle.sync(); + } finally { + await handle.close(); + } + try { + await link(temporary, ownerPath); + await syncDirectory(this.#canonicalRoot!); + const info = await lstat(ownerPath); + this.#ownerIdentity = { device: info.dev, inode: info.ino }; + return; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + const info = await lstat(ownerPath); + if (info.isSymbolicLink() || !info.isFile()) throw new Error("official OMP owner lease is invalid"); + const existing = decodeOwnerRecord(JSON.parse(await readFile(ownerPath, "utf8"))); + if (processIsAlive(existing.pid)) throw new Error("official OMP sessions root already has a live owner"); + const unchanged = await lstat(ownerPath); + if (unchanged.dev !== info.dev || unchanged.ino !== info.ino) + throw new Error("official OMP owner lease changed during recovery"); + await unlink(ownerPath); + await syncDirectory(this.#canonicalRoot!); + } finally { + await unlink(temporary).catch(() => undefined); + } + } + throw new Error("official OMP sessions root owner lease could not be acquired"); + } + + async #assertOwnedDirectory(path: string): Promise { + this.#assertInitialized(); + const info = await lstat(path); + const uid = process.getuid?.(); + if (info.isSymbolicLink() || !info.isDirectory() || (uid !== undefined && info.uid !== uid)) + throw new Error("official OMP session directory is unsafe"); + const canonical = await realpath(path); + const child = relative(this.#canonicalRoot!, canonical); + if (child === "" || child.startsWith("..") || isAbsolute(child)) + throw new Error("official OMP session directory is outside the exclusive root"); + await chmod(canonical, 0o700); + return canonical; + } + + async #persist(): Promise { + const metadata: OfficialProfileMetadata = { version: 1, archived: Object.fromEntries(this.#archived) }; + const body = `${JSON.stringify(metadata)}\n`; + if (Buffer.byteLength(body, "utf8") > METADATA_BYTES) throw new Error("official OMP metadata exceeds 1 MiB"); + const temporary = `${this.#metadataPath}.${Bun.randomUUIDv7()}.tmp`; + try { + const handle = await open(temporary, "wx", 0o600); + try { + await handle.writeFile(body, "utf8"); + await handle.sync(); + } finally { + await handle.close(); + } + await rename(temporary, this.#metadataPath); + await syncDirectory(dirname(this.#metadataPath)); + } catch (error) { + await unlink(temporary).catch(() => undefined); + throw error; + } + } + + async #mutateMetadata(mutation: () => Promise): Promise { + const next = this.#metadataMutation.then(mutation); + this.#metadataMutation = next.catch(() => undefined); + await next; + } +} diff --git a/packages/host-service/src/rpc-child.ts b/packages/host-service/src/rpc-child.ts index b10ade1..0a50a11 100644 --- a/packages/host-service/src/rpc-child.ts +++ b/packages/host-service/src/rpc-child.ts @@ -1,3 +1,4 @@ +import { open, stat } from "node:fs/promises"; import { dirname, resolve } from "node:path"; import { boundedMap, parseBounded, type OperationCapability } from "@t4-code/host-wire"; import type { RpcResponse, RpcSessionEntryFrame } from "./omp-rpc-contract.ts"; @@ -8,6 +9,206 @@ import type { ChildHandle, RpcChildFactory, SessionRecord } from "./types.ts"; const MAX_LINE_BYTES = 1024 * 1024; const STDERR_BYTES = 64 * 1024; const FAILURE_STOP_GRACE_MS = 2_000; +const TRANSCRIPT_READ_BYTES = 64 * 1024; +const MAX_PENDING_DURABLE_CORRELATIONS = 64; + +interface PendingDurableCorrelation { + readonly internalId: string; + readonly message: string; +} + +function rawEntryId(value: unknown): string | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const id = (value as Record).id; + return typeof id === "string" && id.length > 0 && id.length <= 256 ? id : undefined; +} + +function durableEntryId(value: unknown): string | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const entry = value as Record; + if (entry.type === "title" || entry.type === "session") return undefined; + return rawEntryId(entry); +} + +function rawUserMessage( + value: unknown, +): { readonly text: string; readonly entry: Record } | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const entry = value as Record; + if (entry.type !== "message" || !entry.message || typeof entry.message !== "object" || Array.isArray(entry.message)) + return undefined; + const message = entry.message as Record; + if (message.role !== "user") return undefined; + const content = message.content; + if (typeof content === "string") return { text: content, entry }; + if (!Array.isArray(content)) return undefined; + const text = content + .flatMap(part => { + if (typeof part === "string") return [part]; + if (!part || typeof part !== "object" || Array.isArray(part)) return []; + const item = part as Record; + return item.type === "text" && typeof item.text === "string" ? [item.text] : []; + }) + .join(""); + return { text, entry }; +} + +/** + * Reconciles the authoritative OMP JSONL after every RPC frame. Official OMP + * does not publish the fork-only ready watermark or live `session_entry` + * projection, while the released fallback publishes both. Tracking raw entry + * IDs lets one implementation support either runtime without double emission. + */ +class DurableJsonlReconciler { + #offset = 0; + #entryCount = 0; + #lastEntryId: string | null = null; + readonly #projectedEntryIds = new Set(); + readonly #pendingCorrelations: PendingDurableCorrelation[] = []; + #reconcileTail: Promise = Promise.resolve(); + + constructor( + private readonly path: string, + private readonly emit: (frame: RpcSessionEntryFrame) => void, + ) {} + + watermark(): RpcLoadedTranscriptWatermark { + return { lastEntryId: this.#lastEntryId, entryCount: this.#entryCount }; + } + + recordCorrelation(internalId: string, command: Record): void { + if ( + (command.type === "prompt" || command.type === "steer" || command.type === "follow_up") && + typeof command.message === "string" + ) { + if (this.#pendingCorrelations.length >= MAX_PENDING_DURABLE_CORRELATIONS) + throw new Error("too many pending durable prompt correlations"); + this.#pendingCorrelations.push({ internalId, message: command.message }); + } + } + + discardCorrelation(internalId: string): void { + const index = this.#pendingCorrelations.findIndex(item => item.internalId === internalId); + if (index >= 0) this.#pendingCorrelations.splice(index, 1); + } + + observeLiveEntry(value: unknown): Record | undefined { + const id = durableEntryId(value); + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + if (!id) return value as Record; + if (this.#projectedEntryIds.has(id)) return undefined; + this.#projectedEntryIds.add(id); + return this.#correlate(value as Record); + } + + async initialize(): Promise { + let size: number; + try { + const info = await stat(this.path); + if (!info.isFile()) throw new Error("session transcript is not a file"); + size = info.size; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return; + throw error; + } + await this.#scan(size, false); + } + + async reconcile(): Promise { + const next = this.#reconcileTail.then(() => this.#reconcile()); + this.#reconcileTail = next.catch(() => undefined); + await next; + } + + async #reconcile(): Promise { + let info; + try { + info = await stat(this.path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return; + throw error; + } + if (!info.isFile()) throw new Error("session transcript is not a file"); + if (info.size < this.#offset) { + this.#offset = 0; + this.#entryCount = 0; + this.#lastEntryId = null; + this.#projectedEntryIds.clear(); + await this.initialize(); + return; + } + if (info.size === this.#offset) return; + await this.#scan(info.size, true); + } + + async #scan(size: number, publish: boolean): Promise { + const handle = await open(this.path, "r"); + let position = this.#offset; + let pending = Buffer.alloc(0); + try { + while (position < size) { + const length = Math.min(TRANSCRIPT_READ_BYTES, size - position); + const bytes = Buffer.allocUnsafe(length); + const result = await handle.read(bytes, 0, length, position); + if (result.bytesRead === 0) throw new Error("session transcript changed during reconciliation"); + position += result.bytesRead; + const buffered = pending.byteLength === 0 + ? bytes.subarray(0, result.bytesRead) + : Buffer.concat([pending, bytes.subarray(0, result.bytesRead)]); + let start = 0; + let newline = buffered.indexOf(0x0a, start); + while (newline >= 0) { + const line = buffered.subarray(start, newline); + if (line.byteLength > MAX_LINE_BYTES) throw new Error("session transcript line exceeds 1 MiB"); + if (line.byteLength > 0) + this.#observeLine(new TextDecoder("utf-8", { fatal: true }).decode(line), publish); + start = newline + 1; + newline = buffered.indexOf(0x0a, start); + } + pending = Buffer.from(buffered.subarray(start)); + if (pending.byteLength > MAX_LINE_BYTES) throw new Error("session transcript line exceeds 1 MiB"); + } + } finally { + await handle.close(); + } + this.#offset = position - pending.byteLength; + } + + #observeLine(line: string, publish: boolean): void { + let value: unknown; + try { + value = parseBounded(line); + } catch { + throw new Error("malformed session transcript"); + } + const id = durableEntryId(value); + if (!id) return; + this.#entryCount += 1; + this.#lastEntryId = id; + if (!publish || !value || typeof value !== "object" || Array.isArray(value)) return; + if (this.#projectedEntryIds.has(id)) return; + this.#projectedEntryIds.add(id); + this.emit({ type: "session_entry", entry: this.#correlate(value as Record) as never }); + } + + #correlate(entry: Record): Record { + const user = rawUserMessage(entry); + const correlation = this.#pendingCorrelations[0]; + if (user && correlation && user.text === correlation.message) { + this.#pendingCorrelations.shift(); + return { + ...entry, + message: { + ...(entry.message as Record), + ...((entry.message as Record).clientCorrelationId === undefined + ? { clientCorrelationId: correlation.internalId } + : {}), + }, + }; + } + return entry; + } +} export interface ChildCallbacks { entry(frame: RpcSessionEntryFrame): void; @@ -49,8 +250,13 @@ export class BunRpcChildFactory implements RpcChildFactory { #executable: string; #prefixArgv: readonly string[]; #imageRoot: string | undefined; + #environment: Readonly>; - constructor(invocation: string | RpcChildInvocation = resolveRpcChildInvocation(), imageRoot?: string) { + constructor( + invocation: string | RpcChildInvocation = resolveRpcChildInvocation(), + imageRoot?: string, + environment: Readonly> = {}, + ) { const resolved = typeof invocation === "string" ? { executable: invocation, prefixArgv: [] } : invocation; if (typeof resolved.executable !== "string" || resolved.executable.trim().length === 0) { throw new Error("rpc child executable is empty"); @@ -64,6 +270,7 @@ export class BunRpcChildFactory implements RpcChildFactory { this.#executable = resolved.executable; this.#prefixArgv = Object.freeze([...resolved.prefixArgv]); this.#imageRoot = imageRoot; + this.#environment = Object.freeze({ ...environment }); } spawn(spec: { session: SessionRecord; argv: string[]; cwd: string }): ChildHandle { @@ -71,6 +278,7 @@ export class BunRpcChildFactory implements RpcChildFactory { cwd: spec.cwd, env: { ...process.env, + ...this.#environment, OMP_APP_RPC_INLINE_IMAGE_DATA: "omit", OMP_APP_RPC_SESSION_ENTRIES: "1", OMP_APP_SUBAGENT_SUBSCRIPTION: "progress", @@ -152,6 +360,7 @@ export class RpcChildSupervisor { #ready = false; #termination?: Promise; #operationCapabilities: OfficialOmpCapabilityAdapter; + #transcript: DurableJsonlReconciler; constructor( private readonly factory: RpcChildFactory, private readonly session: SessionRecord, @@ -161,6 +370,7 @@ export class RpcChildSupervisor { private readonly runtimeVersion?: string, ) { this.#operationCapabilities = new OfficialOmpCapabilityAdapter(runtimeVersion); + this.#transcript = new DurableJsonlReconciler(session.path, frame => this.callbacks.entry(frame)); if (!Number.isSafeInteger(failureStopGraceMs) || failureStopGraceMs <= 0 || failureStopGraceMs > 60_000) throw new Error("failureStopGraceMs must be between 1 and 60000"); } @@ -202,11 +412,13 @@ export class RpcChildSupervisor { const internalId = `${requestId}:${++this.#counter}`; const promise = Promise.withResolvers(); this.#pending.set(internalId, promise); + this.#transcript.recordCorrelation(internalId, command); const onAbort = () => { const pending = this.#pending.get(internalId); if (!pending) return; this.#pending.delete(internalId); this.#ignoredResponses.add(internalId); + this.#transcript.discardCorrelation(internalId); pending.reject(new Error("rpc call aborted")); if (abortChild) void this.cancel(`${requestId}:cancel`).catch(() => undefined); }; @@ -226,6 +438,7 @@ export class RpcChildSupervisor { return response; } catch (error) { this.#pending.delete(internalId); + this.#transcript.discardCorrelation(internalId); throw error; } finally { signal?.removeEventListener("abort", onAbort); @@ -286,7 +499,12 @@ export class RpcChildSupervisor { // here would let lifecycle retries lose track of a signal-resistant child. } loadedWatermark(): RpcLoadedTranscriptWatermark | undefined { - return this.#loadedWatermark; + return this.#ready ? this.#transcript.watermark() : undefined; + } + async reconcileTranscript(): Promise { + if (!this.#ready || this.#closed) return undefined; + await this.#transcript.reconcile(); + return this.#transcript.watermark(); } child(): ChildHandle | undefined { return this.#child; @@ -325,11 +543,21 @@ export class RpcChildSupervisor { this.#loadedWatermark = { lastEntryId, entryCount }; } if (this.#ready) throw new Error("duplicate rpc ready"); + await this.#transcript.initialize(); + const reconciled = this.#transcript.watermark(); + if ( + this.#loadedWatermark && + (this.#loadedWatermark.entryCount !== reconciled.entryCount || + this.#loadedWatermark.lastEntryId !== reconciled.lastEntryId) + ) + throw new Error("rpc ready watermark does not match durable transcript"); + this.#loadedWatermark = reconciled; this.#ready = true; ready.resolve(); continue; } this.dispatch(frame); + await this.#transcript.reconcile(); } if (!this.#closed) this.fail(new Error("rpc child stdout EOF"), true); } catch (error) { @@ -369,6 +597,15 @@ export class RpcChildSupervisor { throw new Error("rpc response has unknown id"); } this.#pending.delete(value.id); + const responseData = value.data; + const localOnlyPrompt = + value.command === "prompt" && + value.success && + responseData !== null && + typeof responseData === "object" && + !Array.isArray(responseData) && + (responseData as Record).agentInvoked === false; + if (!value.success || localOnlyPrompt) this.#transcript.discardCorrelation(value.id); if (!value.success && typeof value.error !== "string") pending.reject(new Error("rpc response missing error")); else pending.resolve(value as unknown as RpcResponse); return; @@ -376,7 +613,8 @@ export class RpcChildSupervisor { if (value.type === "session_entry") { if (!value.entry || typeof value.entry !== "object" || Array.isArray(value.entry)) throw new Error("malformed rpc session entry"); - this.callbacks.entry(value as unknown as RpcSessionEntryFrame); + const entry = this.#transcript.observeLiveEntry(value.entry); + if (entry) this.callbacks.entry({ ...value, entry } as unknown as RpcSessionEntryFrame); return; } if (typeof value.type !== "string") throw new Error("rpc frame type is missing"); diff --git a/packages/host-service/src/server.ts b/packages/host-service/src/server.ts index e4f488f..680d336 100644 --- a/packages/host-service/src/server.ts +++ b/packages/host-service/src/server.ts @@ -828,6 +828,7 @@ export class LocalAppserver implements AppserverHandle { #partialMarker?: { device: number; inode: number }; #ompVersion: string; #ompBuild: string; + #rpcDialect: NonNullable; #appserverVersion: string; #appserverBuild: string; #supportedFeatures: Set; @@ -835,6 +836,7 @@ export class LocalAppserver implements AppserverHandle { #supportedCapabilities: Set; #projectRootForProject?: AppserverOptions["projectRootForProject"]; #projectRevealer?: AppserverOptions["projectRevealer"]; + #claimLocklessSessions: boolean; #runtimeAdapters?: RuntimeAdapterRegistry; #workspaceAuthority?: WorkspaceAuthority; #workspaceTargetPathForProject?: AppserverOptions["workspaceTargetPathForProject"]; @@ -883,6 +885,7 @@ export class LocalAppserver implements AppserverHandle { this.#transcriptSearch = options.transcriptSearchAuthority; this.#projectRootForProject = options.projectRootForProject; this.#projectRevealer = options.projectRevealer; + this.#claimLocklessSessions = options.claimLocklessSessions === true; this.#runtimeAdapters = options.runtimeAdapters; this.#workspaceAuthority = options.workspaceAuthority; this.#workspaceTargetPathForProject = options.workspaceTargetPathForProject; @@ -893,7 +896,8 @@ export class LocalAppserver implements AppserverHandle { ? new TranscriptImageReader({ root: options.transcriptImageRoot }) : undefined; this.#lockStatus = options.lockStatus ?? (() => "missing"); - this.#factory = options.childFactory ?? new BunRpcChildFactory(options.rpcChildInvocation, this.#imageUploads.root); + this.#factory = options.childFactory ?? + new BunRpcChildFactory(options.rpcChildInvocation, this.#imageUploads.root, options.rpcChildEnvironment); this.#ringSize = options.ringSize ?? 256; if (options.lockStatus && !options.lockCheck) this.#lockCheck = () => { @@ -916,6 +920,7 @@ export class LocalAppserver implements AppserverHandle { throw new Error("usageReadTimeoutMs must be between 1 and 60000"); this.#ompVersion = options.ompVersion ?? "local"; this.#ompBuild = options.ompBuild ?? "local"; + this.#rpcDialect = options.rpcDialect ?? "fork"; this.#baseOperationCapabilities = new OfficialOmpCapabilityAdapter(this.#ompVersion).operations(); this.#appserverVersion = options.appserverVersion ?? "0.1.0"; this.#appserverBuild = options.appserverBuild ?? "local"; @@ -960,11 +965,66 @@ export class LocalAppserver implements AppserverHandle { if (this.#operations?.hasCommand(command)) return true; return ( this.#handlers.has(command) || - DIRECT_SESSION_RPC_COMMANDS.has(command) || + (DIRECT_SESSION_RPC_COMMANDS.has(command) && this.#directRpcCommandSupported(command)) || command === SESSION_CANCEL_COMMAND || command === AGENT_CANCEL_COMMAND ); } + + #directRpcCommandSupported(command: string): boolean { + return ( + this.#rpcDialect === "fork" || + !["session.retry", "session.pause", "session.resume", "session.fast.set"].includes(command) + ); + } + + #directRpcCommand(command: CommandFrame): Record | undefined { + if (!this.#directRpcCommandSupported(command.command)) return undefined; + const type = + command.command === "session.retry" + ? "retry" + : command.command === "session.pause" + ? "pause" + : command.command === "session.resume" + ? "resume" + : command.command === "session.compact" + ? "compact" + : command.command === "session.rename" + ? "set_session_name" + : command.command === "session.model.set" + ? "set_model" + : command.command === "session.thinking.set" + ? "set_thinking_level" + : "set_fast"; + if (command.command === "session.compact") return { type, customInstructions: command.args.instructions }; + if (command.command === "session.rename") return { type, name: command.args.name }; + if (command.command === "session.model.set") { + if (this.#rpcDialect === "official-17.0.6") { + if ( + command.args.persistence !== "session" || + typeof command.args.selector !== "string" || + command.args.role !== undefined + ) + return undefined; + const separator = command.args.selector.indexOf("/"); + if (separator <= 0 || separator === command.args.selector.length - 1) return undefined; + return { + type, + provider: command.args.selector.slice(0, separator), + modelId: command.args.selector.slice(separator + 1), + }; + } + return { + type, + selector: command.args.selector, + role: command.args.role, + persist: command.args.persistence === "settings", + }; + } + if (command.command === "session.thinking.set") return { type, level: command.args.level }; + if (command.command === "session.fast.set") return { type, enabled: command.args.enabled }; + return { type }; + } async start(): Promise { if (this.#started) return; this.#inventoryGeneration += 1; @@ -1866,40 +1926,18 @@ export class LocalAppserver implements AppserverHandle { outcome = { frame: response(this.hostId, command, true, { accepted: true }) }; } } else if (DIRECT_SESSION_RPC_COMMANDS.has(command.command)) { + const rpcCommand = this.#directRpcCommand(command); + if (!rpcCommand) { + outcome = { + frame: response(this.hostId, command, false, undefined, { + code: "unsupported", + message: "command is unavailable in the official OMP RPC runtime", + }), + }; + return this.finish(command, outcome, idempotency); + } const supervisor = await this.ensureSupervisor(command.sessionId!); - const type = - command.command === "session.retry" - ? "retry" - : command.command === "session.pause" - ? "pause" - : command.command === "session.resume" - ? "resume" - : command.command === "session.compact" - ? "compact" - : command.command === "session.rename" - ? "set_session_name" - : command.command === "session.model.set" - ? "set_model" - : command.command === "session.thinking.set" - ? "set_thinking_level" - : "set_fast"; - const args = - command.command === "session.compact" - ? { customInstructions: command.args.instructions } - : command.command === "session.rename" - ? { name: command.args.name } - : command.command === "session.model.set" - ? { - selector: command.args.selector, - role: command.args.role, - persist: command.args.persistence === "settings", - } - : command.command === "session.thinking.set" - ? { level: command.args.level } - : command.command === "session.fast.set" - ? { enabled: command.args.enabled } - : {}; - const result = await supervisor.call({ type, ...args }, command.requestId, controller.signal); + const result = await supervisor.call(rpcCommand, command.requestId, controller.signal); if (!result.success) outcome = { frame: response(this.hostId, command, false, undefined, { @@ -4327,7 +4365,8 @@ export class LocalAppserver implements AppserverHandle { } let observer = this.#observers.get(sessionId); if (!observer) { - const lockless = status === "missing" && !projection.value.ref.liveState?.sessionControl; + const lockless = + !this.#claimLocklessSessions && status === "missing" && !projection.value.ref.liveState?.sessionControl; observer = new SessionTranscriptObserver(record.path, this.hostId); this.#observers.set(sessionId, observer); if (lockless) this.#locklessObservers.add(observer); @@ -4377,7 +4416,26 @@ export class LocalAppserver implements AppserverHandle { this.#promotionFailures.set(sessionId, this.promotionFingerprint(record, projection, poll)); return; } - const final = await observer.poll(); + let final = await observer.poll(); + let loaded = await supervisor.reconcileTranscript(); + for ( + let attempt = 0; + attempt < 4 && + (!final.stable || + final.transcript !== "live" || + final.unresolvedPendingCount !== 0 || + final.watermark.entryCount !== loaded?.entryCount || + final.watermark.lastEntryId !== loaded.lastEntryId); + attempt += 1 + ) { + if (!this.observerIsCurrent(sessionId, observer, record, projection)) { + await this.discardPromotionSupervisor(sessionId, supervisor); + return; + } + await Bun.sleep(10); + final = await observer.poll(); + loaded = await supervisor.reconcileTranscript(); + } if (!this.observerIsCurrent(sessionId, observer, record, projection)) { await this.discardPromotionSupervisor(sessionId, supervisor); return; @@ -4388,7 +4446,6 @@ export class LocalAppserver implements AppserverHandle { return; } await this.applyObserverPoll(sessionId, projection, final); - const loaded = supervisor.loadedWatermark(); if (!this.observerIsCurrent(sessionId, observer, record, projection)) { await this.discardPromotionSupervisor(sessionId, supervisor); return; diff --git a/packages/host-service/src/types.ts b/packages/host-service/src/types.ts index c4c4428..58c5956 100644 --- a/packages/host-service/src/types.ts +++ b/packages/host-service/src/types.ts @@ -243,6 +243,8 @@ export interface AppserverOptions { /** Categorizes external ownership without weakening the write lock gate. */ lockStatus?: SessionLockInspector; /** Final write-lock gate, retained for every child/lifecycle mutation. */ lockCheck?: LockCheckHook; + /** Permit promotion of lockless transcripts only when their whole profile root is exclusively T4-owned. */ + claimLocklessSessions?: boolean; runtimeAdapters?: RuntimeAdapterRegistry; workspaceAuthority?: WorkspaceAuthority; workspaceTargetPathForProject?: (projectId: ProjectId, name: string) => Promise | string; @@ -254,6 +256,10 @@ export interface AppserverOptions { childFactory?: RpcChildFactory; /** OMP RPC executable used when the host owns the generic child factory. */ rpcChildInvocation?: RpcChildInvocation; + /** Bounded profile environment applied only to per-session OMP children. */ + rpcChildEnvironment?: Readonly>; + /** Exact child RPC command dialect; official OMP intentionally exposes a narrower command set. */ + rpcDialect?: "fork" | "official-17.0.6"; appserverVersion?: string; appserverBuild?: string; supportedFeatures?: readonly string[]; diff --git a/packages/host-service/test/hardening.test.ts b/packages/host-service/test/hardening.test.ts index 4916c63..51944f8 100644 --- a/packages/host-service/test/hardening.test.ts +++ b/packages/host-service/test/hardening.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { mkdtemp, stat, writeFile } from "node:fs/promises"; +import { appendFile, mkdtemp, stat, writeFile } from "node:fs/promises"; import { createConnection } from "node:net"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -22,6 +22,7 @@ import { profileSocketPath, unixSocketActive, } from "../src/identity.ts"; +import type { RpcSessionEntryFrame } from "../src/omp-rpc-contract.ts"; import { SessionProjection } from "../src/projection.ts"; import { RpcChildSupervisor, resolveRpcChildInvocation } from "../src/rpc-child.ts"; import { createAppserver } from "../src/server.ts"; @@ -618,6 +619,282 @@ describe("projection, replay, and idempotency", () => { }); describe("child supervision", () => { + test("reconciles official OMP JSONL and conservatively correlates the durable user entry", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-jsonl-")); + const path = join(root, "session.jsonl"); + await writeFile( + path, + [ + JSON.stringify({ type: "session", version: 3, id: "session", timestamp: stamp, cwd: root }), + JSON.stringify({ + type: "message", + id: "existing-user", + message: { role: "user", content: "before" }, + }), + JSON.stringify({ + type: "message", + id: "existing-assistant", + message: { role: "assistant", content: "done" }, + }), + ].join("\n") + "\n", + ); + const written = Promise.withResolvers>(); + const finish = Promise.withResolvers(); + const exited = Promise.withResolvers(); + const child: ChildHandle = { + stdin: { write: data => written.resolve(JSON.parse(data) as Record) }, + stdout: (async function* () { + yield `${JSON.stringify({ type: "ready" })}\n`; + const command = await written.promise; + await appendFile( + path, + `${JSON.stringify({ type: "message", id: "official-user", message: { role: "user", content: command.message } })}\n`, + ); + yield `${JSON.stringify({ type: "response", id: command.id, command: "prompt", success: true })}\n`; + await appendFile( + path, + `${JSON.stringify({ type: "message", id: "official-assistant", message: { role: "assistant", content: "after" } })}\n`, + ); + yield `${JSON.stringify({ type: "message_end", message: { role: "assistant" } })}\n`; + await finish.promise; + })(), + stderr: (async function* () {})(), + exited: exited.promise, + kill: () => { + finish.resolve(); + exited.resolve(0); + }, + }; + const session = { ...record("official-jsonl"), path, cwd: root }; + const entries: Array> = []; + const reconciled = Promise.withResolvers(); + const supervisor = new RpcChildSupervisor( + { + spawn: () => child, + argv: sessionPath => ["omp", "--mode", "rpc", "--session", sessionPath], + }, + session, + { + entry: frame => { + entries.push(frame.entry as Record); + if (entries.length === 2) reconciled.resolve(); + }, + event: () => {}, + crashed: error => reconciled.reject(error), + }, + ); + await supervisor.start(); + expect(supervisor.loadedWatermark()).toEqual({ + lastEntryId: "existing-assistant", + entryCount: 2, + }); + await supervisor.prompt("outer", "after"); + await reconciled.promise; + expect(entries).toEqual([ + { + type: "message", + id: "official-user", + message: { role: "user", content: "after", clientCorrelationId: "outer:1" }, + }, + { + type: "message", + id: "official-assistant", + message: { role: "assistant", content: "after" }, + }, + ]); + expect(supervisor.loadedWatermark()).toEqual({ + lastEntryId: "official-assistant", + entryCount: 4, + }); + supervisor.stop(); + }); + + test("discards local-only prompt correlation before the next durable user entry", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-local-prompt-")); + const path = join(root, "session.jsonl"); + await writeFile( + path, + `${JSON.stringify({ type: "session", version: 3, id: "session", timestamp: stamp, cwd: root })}\n`, + ); + const firstWrite = Promise.withResolvers>(); + const secondWrite = Promise.withResolvers>(); + let writes = 0; + const release = Promise.withResolvers(); + const exited = Promise.withResolvers(); + const child: ChildHandle = { + stdin: { + write: data => { + writes += 1; + (writes === 1 ? firstWrite : secondWrite).resolve(JSON.parse(data) as Record); + }, + }, + stdout: (async function* () { + yield `${JSON.stringify({ type: "ready" })}\n`; + const local = await firstWrite.promise; + yield `${JSON.stringify({ + type: "response", + id: local.id, + command: "prompt", + success: true, + data: { agentInvoked: false }, + })}\n`; + const durable = await secondWrite.promise; + await appendFile( + path, + `${JSON.stringify({ type: "message", id: "durable-user", message: { role: "user", content: durable.message } })}\n`, + ); + yield `${JSON.stringify({ + type: "response", + id: durable.id, + command: "prompt", + success: true, + data: { agentInvoked: true }, + })}\n`; + await release.promise; + })(), + stderr: (async function* () {})(), + exited: exited.promise, + kill: () => { + release.resolve(); + exited.resolve(0); + }, + }; + const entries: Array> = []; + const reconciled = Promise.withResolvers(); + const supervisor = new RpcChildSupervisor( + { + spawn: () => child, + argv: sessionPath => ["omp", "--mode", "rpc", "--session", sessionPath], + }, + { ...record("official-local-prompt"), path, cwd: root }, + { + entry: frame => { + entries.push(frame.entry as Record); + reconciled.resolve(); + }, + event: () => {}, + crashed: error => { throw error; }, + }, + ); + await supervisor.start(); + await supervisor.prompt("local", "/local-only"); + await supervisor.prompt("durable", "real prompt"); + await reconciled.promise; + expect(entries).toHaveLength(1); + expect(entries[0]).toMatchObject({ + id: "durable-user", + message: { role: "user", content: "real prompt", clientCorrelationId: "durable:2" }, + }); + supervisor.stop(); + }); + + test("deduplicates a fork live entry against the same durable JSONL entry", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-fork-jsonl-")); + const path = join(root, "session.jsonl"); + await writeFile( + path, + `${JSON.stringify({ type: "session", version: 3, id: "session", timestamp: stamp, cwd: root })}\n`, + ); + const release = Promise.withResolvers(); + const exited = Promise.withResolvers(); + const entry = { + type: "message", + id: "shared-entry", + message: { role: "assistant", content: "once" }, + }; + const child: ChildHandle = { + stdin: { write: () => {} }, + stdout: (async function* () { + yield `${JSON.stringify({ type: "ready", transcriptWatermark: { lastEntryId: null, entryCount: 0 } })}\n`; + await appendFile(path, `${JSON.stringify(entry)}\n`); + yield `${JSON.stringify({ type: "session_entry", entry })}\n`; + yield `${JSON.stringify({ type: "message_end", message: { role: "assistant" } })}\n`; + await release.promise; + })(), + stderr: (async function* () {})(), + exited: exited.promise, + kill: () => { + release.resolve(); + exited.resolve(0); + }, + }; + const entries: unknown[] = []; + const seen = Promise.withResolvers(); + const settled = Promise.withResolvers(); + const supervisor = new RpcChildSupervisor( + { + spawn: () => child, + argv: sessionPath => ["omp", "--mode", "rpc", "--session", sessionPath], + }, + { ...record("fork-jsonl"), path, cwd: root }, + { + entry: frame => { + entries.push(frame.entry); + seen.resolve(); + }, + event: frame => { + if (frame.type === "message_end") settled.resolve(); + }, + crashed: seen.reject, + }, + ); + await supervisor.start(); + await seen.promise; + await settled.promise; + expect(entries).toEqual([entry]); + expect(supervisor.loadedWatermark()).toEqual({ lastEntryId: "shared-entry", entryCount: 1 }); + supervisor.stop(); + }); + + test("waits for a crash-truncated JSONL record to become durable", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-partial-jsonl-")); + const path = join(root, "session.jsonl"); + const partial = JSON.stringify({ + type: "message", + id: "completed-later", + message: { role: "assistant", content: "complete" }, + }); + const split = Math.floor(partial.length / 2); + await writeFile( + path, + `${JSON.stringify({ type: "session", version: 3, id: "session", timestamp: stamp, cwd: root })}\n${partial.slice(0, split)}`, + ); + const complete = Promise.withResolvers(); + const release = Promise.withResolvers(); + const exited = Promise.withResolvers(); + const child: ChildHandle = { + stdin: { write: () => {} }, + stdout: (async function* () { + yield `${JSON.stringify({ type: "ready" })}\n`; + await complete.promise; + yield `${JSON.stringify({ type: "message_end", message: { role: "assistant" } })}\n`; + await release.promise; + })(), + stderr: (async function* () {})(), + exited: exited.promise, + kill: () => { + release.resolve(); + exited.resolve(0); + }, + }; + const seen = Promise.withResolvers(); + const supervisor = new RpcChildSupervisor( + { + spawn: () => child, + argv: sessionPath => ["omp", "--mode", "rpc", "--session", sessionPath], + }, + { ...record("partial-jsonl"), path, cwd: root }, + { entry: seen.resolve, event: () => {}, crashed: seen.reject }, + ); + await supervisor.start(); + expect(supervisor.loadedWatermark()).toEqual({ lastEntryId: null, entryCount: 0 }); + await appendFile(path, `${partial.slice(split)}\n`); + complete.resolve(); + expect((await seen.promise).entry).toEqual(JSON.parse(partial)); + expect(supervisor.loadedWatermark()).toEqual({ lastEntryId: "completed-later", entryCount: 1 }); + supervisor.stop(); + }); + test("daemon entrypoints resolve RPC children in source and installed layouts", () => { const cases = [ ["/checkout/packages/coding-agent/src/cli/ompd.ts", "/checkout/packages/coding-agent/src/cli.ts"], diff --git a/packages/host-service/test/official-omp-catalog-server.test.ts b/packages/host-service/test/official-omp-catalog-server.test.ts index 11798b0..f5d21cc 100644 --- a/packages/host-service/test/official-omp-catalog-server.test.ts +++ b/packages/host-service/test/official-omp-catalog-server.test.ts @@ -73,6 +73,14 @@ class CapabilityRpcChild implements ChildHandle { ], }, }); + else if (command.type === "set_model") + this.push({ + type: "response", + id, + command: "set_model", + success: true, + data: {}, + }); }, }; @@ -213,6 +221,7 @@ test("attached catalog refresh and terminal-only rejection stay on the runtime b ompVersion: "17.0.6", discovery: { list: async () => [session] }, childFactory: factory, + rpcDialect: "official-17.0.6", lockCheck: () => {}, lockStatus: () => "missing", operationsAuthority: { @@ -228,14 +237,32 @@ test("attached catalog refresh and terminal-only rejection stay on the runtime b protocol: { min: "omp-app/1", max: "omp-app/1" }, client: { name: "live-catalog-test", version: "1", build: "test", platform: "linux" }, requestedFeatures: [], - capabilities: { client: ["sessions.read", "sessions.prompt", "catalog.read"] }, + capabilities: { + client: [ + "sessions.read", + "sessions.prompt", + "sessions.control", + "sessions.manage", + "catalog.read", + ], + }, savedCursors: [], }); expect(await client.nextServer()).toMatchObject({ type: "welcome", - grantedCapabilities: ["sessions.read", "sessions.prompt", "catalog.read"], + grantedCapabilities: [ + "sessions.read", + "sessions.manage", + "sessions.prompt", + "sessions.control", + "catalog.read", + ], }); - expect((await client.nextServer()).type).toBe("sessions"); + const sessionsFrame = await client.nextServer(); + expect(sessionsFrame.type).toBe("sessions"); + if (sessionsFrame.type !== "sessions") throw new Error("session inventory missing"); + const revision = sessionsFrame.sessions[0]?.revision; + if (!revision) throw new Error("session revision missing"); let commandOrdinal = 0; const sendCommand = ( request: string, @@ -291,6 +318,39 @@ test("attached catalog refresh and terminal-only rejection stay on the runtime b const rpcCommandTypes = factory.children[0]?.writes.map((command) => command.type) ?? []; expect(rpcCommandTypes.slice(0, 2)).toEqual(["get_state", "get_available_commands"]); expect(rpcCommandTypes).not.toContain("prompt"); + + sendCommand("retry-stale", "session.retry", {}, { expectedRevision: revision }); + const staleRetry = await responseFor(client, "retry-stale"); + expect(staleRetry).toMatchObject({ ok: false, error: { code: "stale_revision" } }); + if (staleRetry.ok) throw new Error("stale retry unexpectedly succeeded"); + const currentRevision = (staleRetry.error?.details as { actualRevision?: string } | undefined) + ?.actualRevision; + if (!currentRevision) throw new Error("current session revision missing"); + sendCommand("retry", "session.retry", {}, { expectedRevision: currentRevision }); + expect(await responseFor(client, "retry")).toMatchObject({ + ok: false, + error: { + code: "unsupported", + message: "command is unavailable in the official OMP RPC runtime", + }, + }); + expect(factory.children[0]?.writes.map((command) => command.type)).not.toContain("retry"); + + sendCommand( + "model", + "session.model.set", + { selector: "anthropic/claude-sonnet", persistence: "session" }, + { expectedRevision: currentRevision }, + ); + expect(await responseFor(client, "model")).toMatchObject({ ok: true, result: { accepted: true } }); + expect(factory.children[0]?.writes.find((command) => command.type === "set_model")).toMatchObject({ + type: "set_model", + provider: "anthropic", + modelId: "claude-sonnet", + }); + expect(factory.children[0]?.writes.find((command) => command.type === "set_model")).not.toHaveProperty( + "selector", + ); } finally { client.destroy(); await client.closed(); diff --git a/packages/host-service/test/official-omp-profile-authority.test.ts b/packages/host-service/test/official-omp-profile-authority.test.ts new file mode 100644 index 0000000..6a1bbb1 --- /dev/null +++ b/packages/host-service/test/official-omp-profile-authority.test.ts @@ -0,0 +1,174 @@ +import { expect, test } from "bun:test"; +import { mkdir, mkdtemp, readFile, stat, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { OfficialOmpProfileAuthority } from "../src/official-omp-profile-authority.ts"; + +test("isolated official OMP profile authority persists compatible lifecycle state", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-authority-")); + const cwd = join(root, "project"); + const sessionsRoot = join(root, "sessions"); + const metadataPath = join(root, "state", "sessions.json"); + await mkdir(cwd); + const authority = new OfficialOmpProfileAuthority({ sessionsRoot, metadataPath }); + await authority.initialize(); + const created = await authority.create(cwd, "Official session"); + const body = await readFile(created.path, "utf8"); + const newline = body.indexOf("\n"); + expect(Buffer.byteLength(body.slice(0, newline + 1), "utf8")).toBe(256); + expect(JSON.parse(body.slice(0, newline))).toMatchObject({ + type: "title", + v: 1, + title: "Official session", + source: "user", + }); + expect(JSON.parse(body.slice(newline + 1))).toMatchObject({ + type: "session", + version: 3, + id: created.sessionId, + cwd: created.cwd, + }); + const [record] = await authority.list(); + expect(record).toMatchObject({ sessionId: created.sessionId, cwd: created.cwd, title: "Official session" }); + expect(await authority.projectRootForSession(created.sessionId)).toBe(created.cwd); + expect(await authority.projectRootForProject(record!.projectId)).toBe(created.cwd); + + const archivedAt = "2026-07-21T00:00:00.000Z"; + await authority.archive(record!, archivedAt); + await authority.close(); + const restarted = new OfficialOmpProfileAuthority({ sessionsRoot, metadataPath }); + await restarted.initialize(); + const [archived] = await restarted.list(); + expect(archived?.archivedAt).toBe(archivedAt); + await restarted.restore(archived!); + + const artifacts = created.path.slice(0, -".jsonl".length); + await mkdir(artifacts); + await writeFile(join(artifacts, "proof.txt"), "owned"); + await restarted.delete({ ...archived!, archivedAt: undefined }); + await expect(stat(created.path)).rejects.toThrow(); + await expect(stat(artifacts)).rejects.toThrow(); + await restarted.close(); +}); + +test("official OMP profile authority rejects a second live owner for the same root", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-authority-owner-")); + const sessionsRoot = join(root, "sessions"); + const first = new OfficialOmpProfileAuthority({ + sessionsRoot, + metadataPath: join(root, "state-a", "sessions.json"), + }); + const second = new OfficialOmpProfileAuthority({ + sessionsRoot, + metadataPath: join(root, "state-b", "sessions.json"), + }); + await first.initialize(); + await expect(second.initialize()).rejects.toThrow("live owner"); + await first.close(); + await second.initialize(); + await second.close(); +}); + +test("official OMP profile authority serializes concurrent archive metadata", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-authority-metadata-")); + const cwd = join(root, "project"); + const sessionsRoot = join(root, "sessions"); + const metadataPath = join(root, "state", "sessions.json"); + await mkdir(cwd); + const authority = new OfficialOmpProfileAuthority({ sessionsRoot, metadataPath }); + await authority.initialize(); + const first = await authority.create(cwd, "First"); + const second = await authority.create(cwd, "Second"); + const records = await authority.list(); + await Promise.all( + records.map((record, index) => authority.archive(record, `2026-07-21T00:00:0${index}.000Z`)), + ); + const metadata = JSON.parse(await readFile(metadataPath, "utf8")) as { + archived: Record; + }; + expect(Object.keys(metadata.archived).sort()).toEqual([first.sessionId, second.sessionId].sort()); + await authority.close(); +}); + +test("official OMP delete validates artifact ownership before moving the transcript", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-authority-delete-")); + const cwd = join(root, "project"); + await mkdir(cwd); + const authority = new OfficialOmpProfileAuthority({ + sessionsRoot: join(root, "sessions"), + metadataPath: join(root, "state", "sessions.json"), + }); + await authority.initialize(); + const created = await authority.create(cwd); + const [record] = await authority.list(); + await writeFile(created.path.slice(0, -".jsonl".length), "not a directory"); + await expect(authority.delete(record!)).rejects.toThrow("artifact root is unsafe"); + expect((await stat(created.path)).isFile()).toBe(true); + await authority.close(); +}); + +test("official OMP profile authority rejects paths outside its exclusive root", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-authority-boundary-")); + const cwd = join(root, "project"); + const sessionsRoot = join(root, "sessions"); + await mkdir(cwd); + const authority = new OfficialOmpProfileAuthority({ + sessionsRoot, + metadataPath: join(root, "state", "sessions.json"), + }); + await authority.initialize(); + const created = await authority.create(cwd); + const [record] = await authority.list(); + const outside = join(root, "outside.jsonl"); + await writeFile(outside, await readFile(created.path)); + await expect(authority.delete({ ...record!, path: outside })).rejects.toThrow("outside"); + expect((await stat(outside)).isFile()).toBe(true); + await authority.close(); +}); + +test("official OMP profile authority refuses a symlinked session directory", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-authority-symlink-")); + const cwd = join(root, "project"); + const sessionsRoot = join(root, "sessions"); + const outside = join(root, "outside"); + await Promise.all([mkdir(cwd), mkdir(sessionsRoot), mkdir(outside)]); + await symlink(outside, join(sessionsRoot, "-t4")); + const authority = new OfficialOmpProfileAuthority({ + sessionsRoot, + metadataPath: join(root, "state", "sessions.json"), + }); + await authority.initialize(); + await expect(authority.create(cwd)).rejects.toThrow("unsafe"); + expect((await stat(outside)).isDirectory()).toBe(true); + await authority.close(); +}); + +test("official OMP discovery fails closed on symlinked transcript directories and files", async () => { + const root = await mkdtemp(join(tmpdir(), "t4-official-authority-discovery-")); + const cwd = join(root, "project"); + const sessionsRoot = join(root, "sessions"); + const outside = join(root, "outside"); + await Promise.all([mkdir(cwd), mkdir(outside)]); + const authority = new OfficialOmpProfileAuthority({ + sessionsRoot, + metadataPath: join(root, "state", "sessions.json"), + }); + await authority.initialize(); + const created = await authority.create(cwd); + const copied = join(outside, "session-outside.jsonl"); + await writeFile(copied, await readFile(created.path)); + await symlink(outside, join(sessionsRoot, "-outside")); + await expect(authority.list()).rejects.toThrow("symlink"); + await authority.close(); + + const cleanRoot = join(root, "clean-sessions"); + const fileAuthority = new OfficialOmpProfileAuthority({ + sessionsRoot: cleanRoot, + metadataPath: join(root, "state-file", "sessions.json"), + }); + await fileAuthority.initialize(); + await mkdir(join(cleanRoot, "-linked")); + await symlink(copied, join(cleanRoot, "-linked", "session-linked.jsonl")); + await expect(fileAuthority.list()).rejects.toThrow("symlink"); + await fileAuthority.close(); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b0d6bb7..36b7b3d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -332,9 +332,9 @@ importers: '@t4-code/host-service': specifier: workspace:* version: link:../host-service - '@t4-code/host-wire': + '@t4-code/protocol': specifier: workspace:* - version: link:../host-wire + version: link:../protocol devDependencies: '@types/bun': specifier: 1.3.5 diff --git a/scripts/check-release-consistency.mjs b/scripts/check-release-consistency.mjs index 4a5d9ac..7694fc5 100644 --- a/scripts/check-release-consistency.mjs +++ b/scripts/check-release-consistency.mjs @@ -348,6 +348,40 @@ function validateOfficialGate0Snapshot(snapshot, officialRuntime, path, errors) errors.push(`${path} missingOfficialSeams.${seam} must remain explicit`); } } + for (const capability of [ + "jsonlTranscriptReconciliation", + "synthesizedReadyWatermark", + "durableEntryProjection", + "liveEntryDeduplication", + "conservativePromptCorrelation", + ]) { + if (snapshot?.t4AdapterCoverage?.[capability] !== true) { + errors.push(`${path} t4AdapterCoverage.${capability} must be true`); + } + } + if (snapshot?.t4AdapterCoverage?.durableCommandKey !== false) { + errors.push(`${path} t4AdapterCoverage.durableCommandKey must remain false`); + } + if (!isDeepStrictEqual(snapshot?.packagedHostProof?.requiredPlatforms, requiredPlatforms)) { + errors.push(`${path} packagedHostProof.requiredPlatforms must cover macOS ARM64 and Linux x64/ARM64`); + } + if ( + !isDeepStrictEqual(snapshot?.packagedHostProof?.requiredScenarios, [ + "discovery", + "attach", + "prompt", + "durable-jsonl", + "t4-wire-projection", + ]) + ) { + errors.push(`${path} packagedHostProof.requiredScenarios must match the packaged host contract`); + } + if (snapshot?.packagedHostProof?.authorityMode !== "official-exclusive-profile") { + errors.push(`${path} packagedHostProof.authorityMode must be official-exclusive-profile`); + } + if (snapshot?.packagedHostProof?.releasedDefault !== "lycaon-authority-bridge") { + errors.push(`${path} packagedHostProof.releasedDefault must preserve the Lycaon fallback`); + } if (snapshot?.t4Policy?.ambiguousDispatch !== "outcome-unknown-no-auto-replay") { errors.push(`${path} ambiguous dispatch policy must fail closed without automatic replay`); } @@ -913,7 +947,9 @@ export function collectReleaseConsistencyErrors(files, releaseTag) { "official-omp-gate0:", "runner: ubuntu-24.04-arm", "run: pnpm --filter @t4-code/host-service verify:official-omp-lifecycle", - "path: artifacts/official-omp-gate0/${{ matrix.platform }}.json", + "run: pnpm --filter @t4-code/host-daemon verify:official-omp-packaged", + "artifacts/official-omp-gate0/${{ matrix.platform }}.json", + "artifacts/official-omp-packaged-host/${{ matrix.platform }}.json", "tooling:", "cluster:", "actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16", diff --git a/scripts/check-release-consistency.test.mjs b/scripts/check-release-consistency.test.mjs index 4eab710..b049aba 100644 --- a/scripts/check-release-consistency.test.mjs +++ b/scripts/check-release-consistency.test.mjs @@ -106,11 +106,15 @@ test("pins official OMP artifacts and the Gate 0 proof contract", () => { const snapshot = JSON.parse(text); snapshot.runtime.commit = "0".repeat(40); snapshot.requiredScenarios = snapshot.requiredScenarios.filter((item) => item !== "approval"); + snapshot.t4AdapterCoverage.liveEntryDeduplication = false; + snapshot.packagedHostProof.requiredScenarios = ["prompt"]; return JSON.stringify(snapshot); }); const errors = collectReleaseConsistencyErrors(snapshotDrift); assert.ok(errors.some((error) => error.includes("runtime commit must match"))); assert.ok(errors.some((error) => error.includes("requiredScenarios must match"))); + assert.ok(errors.some((error) => error.includes("liveEntryDeduplication must be true"))); + assert.ok(errors.some((error) => error.includes("packagedHostProof.requiredScenarios"))); }); test("rejects a tag that differs from the package version", () => { diff --git a/scripts/ci-paths.mjs b/scripts/ci-paths.mjs index 0d194e1..5f6daa6 100755 --- a/scripts/ci-paths.mjs +++ b/scripts/ci-paths.mjs @@ -33,6 +33,8 @@ const GROUP_PATTERNS = Object.freeze({ /^compat\/(?:official-omp-gate0|omp-app-matrix)\.json$/u, /^docs\/(?:OMP_T4_CAPABILITY_AUDIT\.md|OMP_T4_CAPABILITY_TRACKER\.csv|T4_ARCHITECTURE\.html)$/u, /^packages\/host-service\/(?:bin\/official-omp-gate0\.ts|package\.json)$/u, + /^packages\/host-service\/src\/(?:official-omp-profile-authority|rpc-child|server|types)\.ts$/u, + /^packages\/host-daemon\/(?:bin\/official-omp-packaged-proof\.ts|package\.json|src\/cli\.ts)$/u, /^scripts\/stage-omp-runtime\.mjs$/u, ], tooling: [ diff --git a/scripts/ci-paths.test.mjs b/scripts/ci-paths.test.mjs index 92f0ae6..302558a 100644 --- a/scripts/ci-paths.test.mjs +++ b/scripts/ci-paths.test.mjs @@ -18,6 +18,7 @@ test("host runtime source runs host gates without unrelated platform builds", () ...none, continuity: true, cluster: true, + official_omp_gate0: true, tooling: true, }); }); @@ -73,6 +74,12 @@ test("host daemon changes run the Apple packaging leg", () => { tooling: true, flutter_apple: true, }); + assert.deepEqual(classifyCiPaths(["packages/host-daemon/src/cli.ts"]), { + ...none, + official_omp_gate0: true, + tooling: true, + flutter_apple: true, + }); }); test("mobile web changes run only the Android debug product leg", () => {