From df35ae9c087561fc5edc4639d8716d33a9d483f2 Mon Sep 17 00:00:00 2001 From: bompus Date: Mon, 19 Oct 2026 02:20:07 -0600 Subject: [PATCH 1/2] refactor(t3): remove the t3-v1-steer target and T3 V1 store support Current T3 Code runs only its V2 engine. It keeps its data in statev2.sqlite and has no /api/orchestration/dispatch route or thread.turn.start command, so a t3-v1-steer bridge failed its thread check and could not deliver. - t3StatePath returns only statev2.sqlite; the state.sqlite fallback and the V1 queries in t3ThreadId, t3Threads and automatic wake selection are gone. A statev2.sqlite that does not yet have T3's thread tables (T3 is still creating it) holds no threads; isT3V2 is renamed t3StateReady to say that. - The t3-v1-steer adapter, its snapshot check and the T3_TYPES list are removed. wake-bridge, wake-auto-t3 and credential rotation reject a configuration naming t3-v1-steer with a message that says it was removed and to use t3-v2-queue. - pr-sessions reports a store without thread tables instead of "T3 V1 is unsupported", and its coverage limits no longer mention V1. - Tests that built V1 databases now use the V2 fixtures; the automatic supervisor quarantine and port-follow tests drive a V2 target. Tests of V1-only behavior are deleted. - Docs and the changelog describe the change. --- CHANGELOG.md | 8 + docs/pr-sessions.md | 4 +- docs/usage.md | 6 +- docs/wake-bridges.md | 3 +- scripts/configure-hooks.ts | 4 +- src/pr-sessions.ts | 6 +- src/t3-state.ts | 61 +++---- src/wake-auto-t3.ts | 75 +++----- src/wake-bridge.ts | 10 +- src/wake-credentials.ts | 6 +- src/wake-target.ts | 73 ++------ test/bridge-bridge.test.ts | 141 ++++----------- test/register-hook.test.ts | 39 ++-- test/wake-auto-t3.test.ts | 332 ++++++++++++++++------------------ test/wake-backend.test.ts | 12 +- test/wake-credentials.test.ts | 12 +- test/wake-t3-backend.test.ts | 8 +- 17 files changed, 322 insertions(+), 478 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f83e193..bab405f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ Notable changes to Swarmail. Versions follow [semantic versioning](https://semve ## Unreleased +- The `t3-v1-steer` wake target and support for T3 Code's older database are removed. Current T3 Code runs only its V2 + engine: it keeps its data in `statev2.sqlite` and no longer has the HTTP dispatch route or turn-start command those + bridges sent, so a `t3-v1-steer` bridge failed its thread check and delivered nothing. A `wake-bridge`, `wake-auto-t3` + or credential rotation configuration that still names `t3-v1-steer` now fails at startup with + `target type t3-v1-steer was removed`; change its `target.type` to `t3-v2-queue`. The register hook, `who`, agent + locations, automatic wake and `pr-sessions` read only `statev2.sqlite`. While it is missing, or T3 Code is still + creating it, they find no T3 threads, and `pr-sessions` exits with an error. + - A wake bridge stops with an authentication failure when T3 Code rejects the token its authorization file currently holds (HTTP 401), instead of retrying forever; a standalone bridge whose token was revoked or expired now exits. A 401 for a token that rotation already replaced is still retried, since rotation writes the new header before revoking diff --git a/docs/pr-sessions.md b/docs/pr-sessions.md index 52a1a2b..c872177 100644 --- a/docs/pr-sessions.md +++ b/docs/pr-sessions.md @@ -25,8 +25,8 @@ history and `swarmail.identities`: the swarmail agents whose registration tags n ## Limits -The links are associations, not proof of who implemented the change or released a checkout. Remote environments, T3 V1 -stores, hard-deleted conversations and unlinked sessions are not covered. Only GitHub pull requests are supported: if any +The links are associations, not proof of who implemented the change or released a checkout. Remote environments, +hard-deleted conversations and unlinked sessions are not covered. Only GitHub pull requests are supported: if any conversation in the project links a pull request on another host, such as GitLab, the command fails and exits 1 for every lookup in that project, not only for that pull request. Agent registrations can be overwritten, and historical coverage is incomplete. diff --git a/docs/usage.md b/docs/usage.md index d8b00d2..cce0677 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -369,9 +369,9 @@ returned page read. It works outside a Git checkout and ignores The CLI uses one recognized host session variable, or one registration tied to the current host process's PID and start time. Conflicting variables or several possible sessions fail instead of choosing one. T3 thread identity also matches -registrations made by an earlier provider in that thread. V2 database lookup -requires one matching provider and native ID; V1 uses recorded registration -tags. If discovery fails, +registrations made by an earlier provider in that thread. Looking the thread up +in T3 Code's database requires one matching provider and native ID. If +discovery fails, use the explicit mailbox interface below. Keep registration tags current. `--limit N` sets the page size, capped at 1000, not a total limit. The drain diff --git a/docs/wake-bridges.md b/docs/wake-bridges.md index f3f06fe..7614a8a 100644 --- a/docs/wake-bridges.md +++ b/docs/wake-bridges.md @@ -11,7 +11,6 @@ information to an existing session; it does not assign tasks or manage agents. | Target type | Transport | Delivery | | ------------------- | --------------------------- | --------------------------------------------------------------------------- | -| `t3-v1-steer` | Loopback HTTP | Starts a turn that steers the active turn using its current modes | | `t3-v2-queue` | Loopback HTTP and WebSocket | Agent message with steering intent; can reconcile older queued mail notices | | `opencode-v2-queue` | Loopback HTTP | Prompt with steering delivery and a matching admission receipt | | `codex-queue` | Unix WebSocket | Native app-server steering with a durable command ID | @@ -248,7 +247,7 @@ For a Cursor session that holds a settled automatic T3 registration, `swarmail h The configuration is the [bridge configuration](#configure-and-run) with these differences: - Leave out `target.id`; each thread's bridge supplies its own. -- `target.type` is `t3-v1-steer` or `t3-v2-queue`. `target.delivery` is optional and may only be `auto`, with `t3-v2-queue`. +- `target.type` is `t3-v2-queue`. `target.delivery` is optional and may only be `auto`. - `target.url` is optional. The supervisor reads the port T3 publishes in `/userdata/server-runtime.json` and uses `target.url` only when that file is missing. - `rotation` is required: `verifyBackend` must be `true` and `baseDir` is T3's base directory. It enables the diff --git a/scripts/configure-hooks.ts b/scripts/configure-hooks.ts index 3b27a51..8fe0065 100644 --- a/scripts/configure-hooks.ts +++ b/scripts/configure-hooks.ts @@ -180,8 +180,8 @@ export function withClaudeHooks( /** * The OpenCode plugin: forwards edit-tool calls to `swarmail register` without waiting on it. * One default export serves both majors: OpenCode 2 calls `setup(ctx)`, OpenCode 1 (1.18.29+) - * calls `server(input)`. T3 Code V1 drives OpenCode 1, which shares this plugin directory, - * and OpenCode 2 still discovers the singular `plugin/` directory. + * calls `server(input)`. Both share this plugin directory: OpenCode 2 still discovers the + * singular `plugin/` directory. */ export function openCodePlugin(bin: string): string { return `// Installed by the Swarmail hooks installer (scripts/configure-hooks.ts); edits here are overwritten. diff --git a/src/pr-sessions.ts b/src/pr-sessions.ts index dba901b..f844bf7 100644 --- a/src/pr-sessions.ts +++ b/src/pr-sessions.ts @@ -5,7 +5,7 @@ import { existsSync, realpathSync } from "node:fs"; import { resolve } from "node:path"; import { primaryCheckout } from "./checkout.ts"; import { databasePath, t3Home } from "./paths.ts"; -import { isT3V2, t3StatePath } from "./t3-state.ts"; +import { t3StateReady, t3StatePath } from "./t3-state.ts"; import { identity } from "./tag.ts"; // ACP registry/custom drivers do not identify a native program; only exact manual tags can match them. @@ -586,7 +586,7 @@ export async function prSessions(args: string[]): Promise { requireArgument(existsSync(file), "local T3 store is missing"); const db = new Database(file, { readonly: true }); try { - requireValue(isT3V2(db), "T3 V1 is unsupported"); + requireValue(t3StateReady(db), "local T3 store has no thread tables yet"); db.run("BEGIN"); // Verify the required projection even when no association matches. db.query(`SELECT provider_thread_id,thread_id,owner_node_id,provider_session_id,updated_at,payload_json @@ -617,7 +617,7 @@ export async function prSessions(args: string[]): Promise { database: file, rosterDatabase: room.database, limits: - "Associations are not implementer ownership or checkout release evidence. Remote, V1, non-GitHub, hard-deleted and unlinked sessions are unsupported. Roster identities can be overwritten; historical coverage is incomplete.", + "Associations are not implementer ownership or checkout release evidence. Remote, non-GitHub, hard-deleted and unlinked sessions are unsupported. Roster identities can be overwritten; historical coverage is incomplete.", }, matches, }, diff --git a/src/t3-state.ts b/src/t3-state.ts index be2cfe0..bcd8975 100644 --- a/src/t3-state.ts +++ b/src/t3-state.ts @@ -1,19 +1,17 @@ -// T3 Code's state database. Orchestrator V2 copies V1's `state.sqlite` to `statev2.sqlite` on its -// first start, then writes only the copy and keeps threads in `orchestration_v2_*` tables; the -// copied V1 tables stay behind, frozen. V1 keeps writing `state.sqlite`. +// T3 Code's state database, `statev2.sqlite`. T3 creates it on its first start by copying its +// older database, then keeps threads in `orchestration_v2_*` tables. import { Database } from "bun:sqlite"; import { existsSync } from "node:fs"; import { join } from "node:path"; import { t3Home } from "./paths.ts"; -/** V2's database once a V2 server has created it, else V1's. */ +/** T3's state database under `baseDir`; it may not exist yet. */ export function t3StatePath(baseDir: string): string { - const v2 = join(baseDir, "userdata", "statev2.sqlite"); - return existsSync(v2) ? v2 : join(baseDir, "userdata", "state.sqlite"); + return join(baseDir, "userdata", "statev2.sqlite"); } -/** Whether `db` holds V2's thread tables. */ -export function isT3V2(db: Database): boolean { +/** Whether `db` has its thread tables yet; while T3 is still creating it, it holds no threads. */ +export function t3StateReady(db: Database): boolean { return ( db .query( @@ -26,7 +24,7 @@ export function isT3V2(db: Database): boolean { /** * V2's app threads with their current root provider thread. `session` is the provider's own * session id, `status` is `running` while that provider thread is loaded (V2 `idle` or `active`), - * as V1's runtime rows reported it, and `last_seen_at` is its last update. + * and `last_seen_at` is its last update. */ export const T3_V2_THREADS = ` select t.thread_id, t.title, t.project_id, t.archived_at, t.deleted_at, @@ -43,8 +41,8 @@ export const T3_V2_THREADS = ` /** * The T3 Code thread that runs provider session `sessionId`, or null outside T3. T3 passes no - * thread id to the provider process, but V1's resume cursor for the thread and V2's native - * thread reference hold the session id. + * thread id to the provider process, but the provider thread's native reference holds the + * session id. With `host`, the provider must match too. */ export function t3ThreadId( sessionId: string, @@ -57,11 +55,10 @@ export function t3ThreadId( try { const db = new Database(dbPath, { readonly: true }); try { + if (!t3StateReady(db)) { + return null; + } if (host !== undefined) { - // V1 resume cursors do not establish the native provider. Use the registration tag there. - if (!isT3V2(db)) { - return null; - } const driver = host === "claude" ? "claudeAgent" : host === "agy" ? "antigravity" : host; const rows = db .query<{ thread_id: string }, [string, string]>( @@ -73,17 +70,11 @@ export function t3ThreadId( } return rows[0]?.thread_id ?? null; } - const row = isT3V2(db) - ? db - .query<{ thread_id: string }, [string]>( - "select thread_id from orchestration_v2_projection_provider_threads where thread_id is not null and json_extract(payload_json, '$.nativeThreadRef.nativeId') = ? order by updated_at desc limit 1", - ) - .get(sessionId) - : db - .query<{ thread_id: string }, [string]>( - "select thread_id from provider_session_runtime where instr(resume_cursor_json, ?) > 0 order by last_seen_at desc limit 1", - ) - .get(JSON.stringify(sessionId)); + const row = db + .query<{ thread_id: string }, [string]>( + "select thread_id from orchestration_v2_projection_provider_threads where thread_id is not null and json_extract(payload_json, '$.nativeThreadRef.nativeId') = ? order by updated_at desc limit 1", + ) + .get(sessionId); return row?.thread_id ?? null; } finally { db.close(); @@ -104,7 +95,7 @@ export interface T3Thread { last_seen_at: string | null; } -/** T3's threads by id; none while the database is missing or mid-copy on V2's first start. */ +/** T3's threads by id; none while the database is missing or T3 is still creating it. */ export function t3Threads(dbPath: string): Map { if (!existsSync(dbPath)) { return new Map(); @@ -116,19 +107,13 @@ export function t3Threads(dbPath: string): Map { return new Map(); } try { + if (!t3StateReady(db)) { + return new Map(); + } const rows = db .query( - isT3V2(db) - ? ` - select thread_id, title, cwd, status, last_seen_at from (${T3_V2_THREADS}) - where deleted_at is null` - : ` - select t.thread_id, t.title, coalesce(t.worktree_path, p.workspace_root) as cwd, - r.status, r.last_seen_at - from projection_threads t - left join projection_projects p on p.project_id = t.project_id - left join provider_session_runtime r on r.thread_id = t.thread_id - where t.deleted_at is null`, + `select thread_id, title, cwd, status, last_seen_at from (${T3_V2_THREADS}) + where deleted_at is null`, ) .all(); return new Map(rows.map((row) => [row.thread_id, row])); diff --git a/src/wake-auto-t3.ts b/src/wake-auto-t3.ts index 1166ae7..30f969c 100644 --- a/src/wake-auto-t3.ts +++ b/src/wake-auto-t3.ts @@ -2,13 +2,13 @@ import { Database } from "bun:sqlite"; import { mkdirSync, readFileSync, statSync } from "node:fs"; import { resolve } from "node:path"; -import { isT3V2, T3_V2_THREADS, t3StatePath } from "./t3-state.ts"; +import { t3StateReady, T3_V2_THREADS, t3StatePath } from "./t3-state.ts"; import { checkT3Target, BridgeError, TargetAuthenticationError, localUrl, - T3_TYPES, + rejectRemovedTargetType, } from "./wake-target.ts"; import type { WakeTarget } from "./wake-target.ts"; import { primaryCheckout } from "./checkout.ts"; @@ -38,8 +38,7 @@ type AutoState = RegisterState & { wakeThread?: string }; * Providers whose T3 V2 threads this service wakes. Claude is left out because Claude in T3 * already waits on its own session id (the wake mod, or the stop hook without it). A bridge would * wait on the same id, and the server keeps one waiter per id, so each would replace the other - * with a 409. Under T3's Cursor SDK, Cursor's stop hook stops waiting when the turn ends. V1 stays - * Codex-only. + * with a 409. Under T3's Cursor SDK, Cursor's stop hook stops waiting when the turn ends. */ const WAKE_PROVIDERS = "('codex','cursor')"; @@ -147,37 +146,28 @@ export function enrollT3Threads( ): void { const states = openRegistry(registry); const rows = readT3((db) => - db - .query< - { - thread_id: string; - session: unknown; - cwd: string; - last_seen_at: string; - provider_name: string; - }, - [string] - >( - isT3V2(db) - ? ` + t3StateReady(db) + ? db + .query< + { + thread_id: string; + session: unknown; + cwd: string; + last_seen_at: string; + provider_name: string; + }, + [string] + >( + ` SELECT thread_id, session, last_seen_at, cwd, provider_name FROM (${T3_V2_THREADS}) v WHERE deleted_at IS NULL AND archived_at IS NULL AND NOT coalesce(subagent, 0) AND ${UNSETTLED_V2} AND cwd IS NOT NULL AND provider_name IN ${WAKE_PROVIDERS} AND status='running' AND julianday(last_seen_at) >= julianday(?) - ` - : ` - SELECT t.thread_id, CASE WHEN json_valid(r.resume_cursor_json) - THEN json_extract(r.resume_cursor_json,'$.threadId') END AS session, r.last_seen_at, - coalesce(t.worktree_path,p.workspace_root) AS cwd, 'codex' AS provider_name - FROM projection_threads t JOIN provider_session_runtime r ON r.thread_id=t.thread_id - JOIN projection_projects p ON p.project_id=t.project_id - WHERE t.deleted_at IS NULL AND t.archived_at IS NULL - AND t.settled_at IS NULL AND t.settled_override IS NOT 'settled' - AND r.provider_name='codex' AND r.status='running' AND julianday(r.last_seen_at) >= julianday(?) `, - ) - .all(after), + ) + .all(after) + : [], ); for (const row of rows) { const session = row.session; @@ -229,25 +219,18 @@ export function automaticT3Threads( ): Set { const states = openRegistry(registry); const rows = readT3((db) => - db - .query<{ thread_id: string; session: unknown; provider_name: string }, []>( - isT3V2(db) - ? ` + t3StateReady(db) + ? db + .query<{ thread_id: string; session: unknown; provider_name: string }, []>( + ` SELECT thread_id, session, provider_name FROM (${T3_V2_THREADS}) v WHERE deleted_at IS NULL AND archived_at IS NULL AND NOT coalesce(subagent, 0) AND ${UNSETTLED_V2} AND provider_name IN ${WAKE_PROVIDERS} - ` - : ` - SELECT t.thread_id, CASE WHEN json_valid(r.resume_cursor_json) - THEN json_extract(r.resume_cursor_json,'$.threadId') END AS session, - 'codex' AS provider_name - FROM projection_threads t JOIN provider_session_runtime r ON r.thread_id=t.thread_id - WHERE t.deleted_at IS NULL AND t.archived_at IS NULL - AND t.settled_at IS NULL AND t.settled_override IS NOT 'settled' AND r.provider_name='codex' `, - ) - .all(), + ) + .all() + : [], ); const threads = new Set(); // Only the registration named by each thread's provider session can match it. @@ -284,10 +267,10 @@ export function automaticT3Threads( export function automaticT3Configuration(path: string) { const config = JSON.parse(readFileSync(path, "utf8")); + rejectRemovedTargetType(config.target?.type); if ( - !T3_TYPES.includes(config.target?.type) || - (config.target.delivery !== undefined && - (config.target.type !== "t3-v2-queue" || config.target.delivery !== "auto")) || + config.target?.type !== "t3-v2-queue" || + (config.target.delivery !== undefined && config.target.delivery !== "auto") || config.rotation?.verifyBackend !== true || typeof config.rotation?.baseDir !== "string" || typeof config.enrollAfter !== "string" || diff --git a/src/wake-bridge.ts b/src/wake-bridge.ts index d81aa57..dca80f8 100644 --- a/src/wake-bridge.ts +++ b/src/wake-bridge.ts @@ -2,7 +2,13 @@ // SQLite holds the pending command and an owner lock released when its connection closes. import { ensureWakeEligible } from "./wake-lifecycle.ts"; import { readFileSync } from "node:fs"; -import { BridgeError, localUrl, openCodeAdapter, t3V1Adapter, t3V2Adapter } from "./wake-target.ts"; +import { + BridgeError, + localUrl, + openCodeAdapter, + rejectRemovedTargetType, + t3V2Adapter, +} from "./wake-target.ts"; import type { TargetAdapter, WakeTarget } from "./wake-target.ts"; import { grokAdapter } from "./wake-grok.ts"; import type { GrokTarget } from "./wake-grok.ts"; @@ -20,7 +26,6 @@ type Target = WakeTarget | GrokTarget | CodexTarget; // The HTTP types share WakeTarget, whose `type` is itself a union. type TargetOf = K extends WakeTarget["type"] ? WakeTarget : Extract; const ADAPTERS: { [K in Target["type"]]: TargetAdapter> } = { - "t3-v1-steer": t3V1Adapter, "t3-v2-queue": t3V2Adapter, "opencode-v2-queue": openCodeAdapter, "grok-queue": grokAdapter, @@ -45,6 +50,7 @@ function readConfig(path: string): BridgeConfig { throw new BridgeError("config requires a target session/thread id"); } const swarmailUrl = localUrl(input.swarmailUrl ?? "http://127.0.0.1:18765"); + rejectRemovedTargetType(target.type); const adapter = adapterFor(target.type); if (!adapter) { throw new BridgeError("config requires target type and absolute authorizationFile"); diff --git a/src/wake-credentials.ts b/src/wake-credentials.ts index 63181c6..6a1f9b0 100644 --- a/src/wake-credentials.ts +++ b/src/wake-credentials.ts @@ -15,7 +15,7 @@ import { writeFileSync, } from "node:fs"; import { dirname, isAbsolute, join } from "node:path"; -import { BridgeError, localUrl, T3_TYPES } from "./wake-target.ts"; +import { BridgeError, localUrl, rejectRemovedTargetType } from "./wake-target.ts"; import { backendTimeout, checkBackendOperation, @@ -87,9 +87,9 @@ function readConfig(path: string) { const config = JSON.parse(readFileSync(path, "utf8")); const target = config.target; const rotation = config.rotation; + rejectRemovedTargetType(target?.type); if ( - !target || - !T3_TYPES.includes(target.type) || + target?.type !== "t3-v2-queue" || (target.id !== undefined && (typeof target.id !== "string" || !target.id)) || !rotation ) { diff --git a/src/wake-target.ts b/src/wake-target.ts index 1cabbe6..efcd60a 100644 --- a/src/wake-target.ts +++ b/src/wake-target.ts @@ -3,7 +3,7 @@ import { readFileSync, statSync } from "node:fs"; import { isAbsolute } from "node:path"; export interface WakeTarget { - type: "t3-v1-steer" | "t3-v2-queue" | "opencode-v2-queue"; + type: "t3-v2-queue" | "opencode-v2-queue"; url: string; id: string; authorizationFile: string; @@ -21,8 +21,14 @@ export class BridgeError extends Error { /** Authentication is shared by every thread using the configured target. */ export class TargetAuthenticationError extends BridgeError {} -/** The target types that reach a T3 thread. */ -export const T3_TYPES: readonly string[] = ["t3-v1-steer", "t3-v2-queue"]; +/** Fails loudly on a target type Swarmail no longer delivers to, instead of reporting it as unknown. */ +export function rejectRemovedTargetType(type: unknown): void { + if (type === "t3-v1-steer") { + throw new BridgeError( + "target type t3-v1-steer was removed: current T3 Code runs only its V2 engine, which has no V1 route; use t3-v2-queue", + ); + } +} /** One wake target type: its config, the destination its state binds to, and its protocol. */ export interface TargetAdapter { @@ -135,32 +141,6 @@ async function request(target: WakeTarget, path: string, body?: unknown): Promis } } -/** Shared snapshot validation for delivery and the read-only installed-target check. */ -async function inspectT3Thread(target: WakeTarget) { - const snapshot = await request( - target, - `/api/orchestration/threads/${encodeURIComponent(target.id)}`, - ); - if (target.type === "t3-v2-queue") { - if ( - snapshot?.projection?.thread?.id !== target.id || - !Array.isArray(snapshot.projection.runs) - ) { - throw new BridgeError("target does not expose the tested T3 V2 projection protocol"); - } - return snapshot.projection.thread; - } - const thread = snapshot?.thread; - if ( - thread?.id !== target.id || - !["full-access", "approval-required"].includes(thread.runtimeMode) || - !["default", "plan"].includes(thread.interactionMode) - ) { - throw new BridgeError("target does not expose the tested T3 V1 thread protocol"); - } - return thread; -} - /** Shared GET-only auth check, including automatic supervisors with no eligible thread yet. */ export async function checkT3Authentication(target: WakeTarget): Promise { let session; @@ -185,11 +165,17 @@ export async function checkT3Authentication(target: WakeTarget): Promise { /** GET-only preflight; proves auth/snapshot compatibility, never dispatch or model execution. */ export async function checkT3Target(target: WakeTarget): Promise { - if (!T3_TYPES.includes(target.type)) { + if (target.type !== "t3-v2-queue") { throw new BridgeError("health check requires a T3 target"); } await checkT3Authentication(target); - await inspectT3Thread(target); + const snapshot = await request( + target, + `/api/orchestration/threads/${encodeURIComponent(target.id)}`, + ); + if (snapshot?.projection?.thread?.id !== target.id || !Array.isArray(snapshot.projection.runs)) { + throw new BridgeError("target does not expose the tested T3 V2 projection protocol"); + } } async function dispatchV2( @@ -514,33 +500,10 @@ function admitted(result: any) { // Each prepared command is journaled whole, so a retry never rereads mutable thread settings. -/** T3 V1: start a turn that steers the active one, with the thread's current modes. */ -export const t3V1Adapter: TargetAdapter = { - parse: parseHttp("t3-v1-steer"), - // The port moves on T3 restart; the authorization file ties the bridge to one T3 database. - binding: () => ({}), - prepare: async (target, text) => { - const thread = await inspectT3Thread(target); - return { - type: "thread.turn.start", - commandId: crypto.randomUUID(), - threadId: target.id, - message: { messageId: crypto.randomUUID(), role: "user", text, attachments: [] }, - runtimeMode: thread.runtimeMode, - interactionMode: thread.interactionMode, - createdAt: new Date().toISOString(), - }; - }, - deliver: async (target, command, _attempted, _signal, guard) => { - ensureT3Unsettled(await inspectT3Thread(target)); - await guard?.(); - admitted(await request(target, "/api/orchestration/dispatch", command)); - }, -}; - /** T3 V2: steer the active run, or start an idle thread. */ export const t3V2Adapter: TargetAdapter = { parse: parseHttp("t3-v2-queue"), + // The port moves on T3 restart; the authorization file ties the bridge to one T3 database. binding: () => ({}), prepare: () => { throw new BridgeError("T3 V2 notices require the notice adapter"); diff --git a/test/bridge-bridge.test.ts b/test/bridge-bridge.test.ts index 19169fb..3b6e313 100644 --- a/test/bridge-bridge.test.ts +++ b/test/bridge-bridge.test.ts @@ -5,7 +5,7 @@ import { Database } from "bun:sqlite"; import { createServer as createTcpServer, type AddressInfo, type Socket } from "node:net"; import { createServer } from "../src/server.ts"; import { openWakeState, wakeStatePath } from "../src/wake-state.ts"; -import { BridgeError, t3V1Adapter, t3V2Adapter, type WakeTarget } from "../src/wake-target.ts"; +import { BridgeError, t3V2Adapter, type WakeTarget } from "../src/wake-target.ts"; import { bridgeBinding } from "../src/wake-bridge.ts"; import { testScratch } from "./fixtures/test-scratch.ts"; import type { Args, Row } from "../src/store.ts"; @@ -42,7 +42,6 @@ async function until(fn: () => unknown) { /** How the fake target answers. */ interface Behavior { reject?: boolean; - settled?: boolean; mismatch?: boolean; deliveryMismatch?: boolean; } @@ -60,7 +59,6 @@ async function fixture(type = "opencode-v2-queue", behavior: Behavior = {}) { const received: any[] = []; const authorizations: (string | null)[] = []; let reject = behavior.reject ?? false; - let settled = behavior.settled ?? false; const target = Bun.serve({ hostname: "127.0.0.1", port: 0, @@ -74,18 +72,10 @@ async function fixture(type = "opencode-v2-queue", behavior: Behavior = {}) { return Response.json({ ticket: "private-ticket" }); } if (req.method === "GET") { - const thread = { - id, - runtimeMode: "approval-required", - interactionMode: "plan", - settledAt: null, - settledOverride: settled ? "settled" : null, - }; - return Response.json( - type === "t3-v1-steer" - ? { thread } - : { projection: { thread, runs: [], messages: [], runtimeRequests: [] } }, - ); + const thread = { id, settledAt: null, settledOverride: null }; + return Response.json({ + projection: { thread, runs: [], messages: [], runtimeRequests: [] }, + }); } const body = (await req.json()) as Args; received.push(body); @@ -221,12 +211,6 @@ async function fixture(type = "opencode-v2-queue", behavior: Behavior = {}) { allow: () => { reject = false; }, - settle: () => { - settled = true; - }, - reactivate: () => { - settled = false; - }, }; } @@ -334,38 +318,38 @@ test("mismatched admission keeps mail pending and never prints credentials or pr expect(await new Response(run.stderr).text()).toContain("does not match"); }); -for (const type of ["t3-v1-steer", "t3-v2-queue"]) { - test(`${type} uses its native protocol and retains the thread's permissions`, async () => { - const f = await fixture(type); - await f.send(); - const run = f.start(); - await until(() => f.state()?.acknowledged > 0); - run.kill(); - await run.exited; - if (type === "t3-v1-steer") { - expect(f.received[0]).toMatchObject({ - type: "thread.turn.start", - threadId: f.id, - runtimeMode: "approval-required", - interactionMode: "plan", - }); - expect(f.received[0]).not.toHaveProperty("modelSelection"); - } else { - expect(f.received[0]).toMatchObject({ - _tag: "Request", - tag: "orchestration.dispatchCommand", - payload: { - type: "message.dispatch", - threadId: f.id, - deliveryIntent: "steer", - dispatchMode: { type: "start_immediately" }, - createdBy: "agent", - }, - }); - expect(f.received[0].payload).not.toHaveProperty("modelSelection"); - } +test("t3-v2-queue uses its native protocol and keeps the thread's model", async () => { + const f = await fixture("t3-v2-queue"); + await f.send(); + const run = f.start(); + await until(() => f.state()?.acknowledged > 0); + run.kill(); + await run.exited; + expect(f.received[0]).toMatchObject({ + _tag: "Request", + tag: "orchestration.dispatchCommand", + payload: { + type: "message.dispatch", + threadId: f.id, + deliveryIntent: "steer", + dispatchMode: { type: "start_immediately" }, + createdBy: "agent", + }, }); -} + expect(f.received[0].payload).not.toHaveProperty("modelSelection"); +}); + +test("a t3-v1-steer configuration fails at load and names its replacement", async () => { + const f = await fixture("t3-v1-steer"); + await f.send(); + const run = f.start(); + expect(await run.exited).toBe(1); + expect(await new Response(run.stderr).text()).toContain( + "target type t3-v1-steer was removed: current T3 Code runs only its V2 engine, which has no V1 route; use t3-v2-queue", + ); + expect(f.received).toHaveLength(0); + expect(f.state()).toBeNull(); +}); test("rejects destination changes with pending mail", async () => { const f = await fixture("opencode-v2-queue", { mismatch: true }); @@ -532,7 +516,7 @@ test("two bridges in one process cannot own the same mailbox session", async () } rmSync(dir, { recursive: true, force: true }); }); - const binding = JSON.stringify({ type: "t3-v1-steer", id: "same-thread" }); + const binding = JSON.stringify({ type: "t3-v2-queue", id: "same-thread" }); const open = () => openWakeState("http://127.0.0.1:1", "same-thread", binding); const first = open(); expect(open).toThrow("another wake bridge owns this mailbox session"); @@ -551,7 +535,6 @@ test("each target type keeps its saved binding", () => { expect( ( [ - { type: "t3-v1-steer", ...http }, { type: "t3-v2-queue", ...http }, { type: "opencode-v2-queue", ...http }, { type: "grok-queue", ...native }, @@ -559,7 +542,6 @@ test("each target type keeps its saved binding", () => { ] as const ).map(bridgeBinding), ).toEqual([ - '{"type":"t3-v1-steer","id":"t"}', '{"type":"t3-v2-queue","id":"t"}', '{"url":"http://127.0.0.1:1","type":"opencode-v2-queue","id":"t"}', '{"socket":"/s","cwd":"/c","type":"grok-queue","id":"t"}', @@ -567,55 +549,6 @@ test("each target type keeps its saved binding", () => { ]); }); -test("V1 delivery holds an immutable command while settled and resumes after bridge restart", async () => { - const f = await fixture("t3-v1-steer"); - await f.send(); - const config = JSON.parse(readFileSync(f.config, "utf8")); - const previous = process.env.XDG_STATE_HOME; - process.env.XDG_STATE_HOME = join(f.dir, "state"); - const open = () => openWakeState(config.swarmailUrl, f.id, "{}"); - let journal = open(); - try { - journal.savePending({ - eventId: 1, - command: await t3V1Adapter.prepare(config.target, "Swarmail: pending mail"), - }); - const saved = structuredClone(journal.pending!); - f.settle(); // Settlement follows preparation, before the native dispatch. - journal.close(); - journal = open(); - expect( - t3V1Adapter.deliver( - config.target, - journal.pending!.command, - () => journal.markAttempted(), - new AbortController().signal, - ), - ).rejects.toMatchObject({ retryable: true, message: "T3 thread is settled; holding mail" }); - expect(f.received).toHaveLength(0); - expect(journal.acknowledged).toBe(0); - expect(journal.pending).toEqual(saved); - f.reactivate(); - await t3V1Adapter.deliver( - config.target, - journal.pending!.command, - () => journal.markAttempted(), - new AbortController().signal, - ); - journal.accept(saved.eventId); - expect(f.received).toEqual([saved.command]); - expect(journal.pending).toBeNull(); - expect( - f.mail.db - .query("SELECT count(*) AS unread FROM message_recipients WHERE read_ts IS NULL") - .get()!.unread, - ).toBe(1); - } finally { - journal.close(); - process.env.XDG_STATE_HOME = previous; - } -}); - test("OpenCode receipt must match the saved steering mode", async () => { const f = await fixture("opencode-v2-queue", { deliveryMismatch: true }); await f.send(); diff --git a/test/register-hook.test.ts b/test/register-hook.test.ts index 62fc8c0..1bd2d26 100644 --- a/test/register-hook.test.ts +++ b/test/register-hook.test.ts @@ -2,7 +2,6 @@ import { expect, test } from "bun:test"; import { Database } from "bun:sqlite"; import { realpathSync, - copyFileSync, existsSync, mkdirSync, mkdtempSync, @@ -25,7 +24,7 @@ import { withLock, } from "../src/registry.ts"; import type { Register, RegisterState } from "../src/registry.ts"; -import { t3StatePath, t3ThreadId } from "../src/t3-state.ts"; +import { t3StatePath, t3ThreadId, t3Threads } from "../src/t3-state.ts"; import { addT3V2Thread, createT3V2Tables } from "./fixtures/t3-v2-state.ts"; import { parseTag, sessionTag } from "../src/tag.ts"; import { createServer } from "../src/server.ts"; @@ -384,16 +383,15 @@ test("tells Claude hook input from Cursor, Devin and Grok input", () => { }); }); -test("finds the T3 thread whose resume cursor holds the session id", () => { +test("finds the T3 thread whose native thread reference holds the session id", () => { const dir = realpathSync.native(mkdtempSync(join(tmpdir(), "t3-state-"))); - const path = join(dir, "state.sqlite"); + mkdirSync(join(dir, "userdata")); + const path = t3StatePath(dir); + expect(path).toBe(join(dir, "userdata/statev2.sqlite")); const db = new Database(path); - db.run( - "create table provider_session_runtime (thread_id text, resume_cursor_json text, last_seen_at text)", - ); - db.run( - "insert into provider_session_runtime values ('thread-a', '{\"resume\":\"sess-1\"}', '1'), ('thread-b', '{\"sessionId\":\"sess-2\"}', '2')", - ); + createT3V2Tables(db); + addT3V2Thread(db, { threadId: "thread-a", nativeId: "sess-1" }); + addT3V2Thread(db, { threadId: "thread-b", nativeId: "sess-2" }); db.close(); try { expect(t3ThreadId("sess-1", path)).toBe("thread-a"); @@ -405,11 +403,12 @@ test("finds the T3 thread whose resume cursor holds the session id", () => { } }); -test("after the V2 cutover, finds the thread from V2's native thread reference", () => { +test("a T3 database still being created holds no threads", () => { const dir = realpathSync.native(mkdtempSync(join(tmpdir(), "t3-state-"))); mkdirSync(join(dir, "userdata")); - const v1 = join(dir, "userdata/state.sqlite"); - const db = new Database(v1); + const path = t3StatePath(dir); + // T3 creates its database by copying its older one, whose tables arrive before its own. + const db = new Database(path); db.run( "create table provider_session_runtime (thread_id text, resume_cursor_json text, last_seen_at text)", ); @@ -418,17 +417,9 @@ test("after the V2 cutover, finds the thread from V2's native thread reference", ); db.close(); try { - expect(t3StatePath(dir)).toBe(v1); - // V2 starts from a copy of V1's database and leaves the copied runtime rows frozen. - const v2 = join(dir, "userdata/statev2.sqlite"); - copyFileSync(v1, v2); - const copy = new Database(v2); - createT3V2Tables(copy); - addT3V2Thread(copy, { threadId: "thread-b", nativeId: "sess-2" }); - copy.close(); - expect(t3StatePath(dir)).toBe(v2); - expect(t3ThreadId("sess-2", v2)).toBe("thread-b"); - expect(t3ThreadId("sess-1", v2)).toBeNull(); + expect(t3ThreadId("sess-1", path)).toBeNull(); + expect(t3ThreadId("sess-1", path, "codex")).toBeNull(); + expect(t3Threads(path).size).toBe(0); } finally { rmSync(dir, { recursive: true, force: true }); } diff --git a/test/wake-auto-t3.test.ts b/test/wake-auto-t3.test.ts index 993f7fd..89bcd11 100644 --- a/test/wake-auto-t3.test.ts +++ b/test/wake-auto-t3.test.ts @@ -36,22 +36,15 @@ const test = process.platform === "linux" ? bunTest : bunTest.skip; const scratch = testScratch(); +/** T3's database with one running Codex thread, `thread-1` on native session `native-1`. */ function fixture() { const root = mkdtempSync(join(scratch, "case-")); const registry = join(root, "state/swarmail-register"); mkdirSync(registry, { recursive: true }); mkdirSync(join(root, ".t3/userdata"), { recursive: true }); - const db = new Database(join(root, ".t3/userdata/state.sqlite")); - db.exec(`CREATE TABLE projection_threads(thread_id TEXT PRIMARY KEY, deleted_at TEXT, archived_at TEXT, project_id TEXT, worktree_path TEXT, settled_at TEXT, settled_override TEXT); - CREATE TABLE projection_projects(project_id TEXT PRIMARY KEY, workspace_root TEXT); - CREATE TABLE provider_session_runtime(thread_id TEXT PRIMARY KEY, provider_name TEXT, resume_cursor_json TEXT, last_seen_at TEXT, status TEXT);`); - db.query("INSERT INTO projection_threads VALUES (?,NULL,NULL,'project-1',NULL,NULL,NULL)").run( - "thread-1", - ); - db.query("INSERT INTO provider_session_runtime VALUES (?,'codex',?,'2026-09-30','running')").run( - "thread-1", - JSON.stringify({ threadId: "native-1" }), - ); + const db = new Database(join(root, ".t3/userdata/statev2.sqlite")); + createT3V2Tables(db); + addT3V2Thread(db, { threadId: "thread-1", nativeId: "native-1", updatedAt: "2026-09-30" }); return { root, registry, @@ -64,6 +57,20 @@ function fixture() { }; } +/** Sets a T3 thread's settlement override; null reactivates it. Without `thread`, every thread's. */ +function settleT3(db: Database, override: string | null, thread?: string) { + db.query( + `UPDATE orchestration_v2_projection_threads SET payload_json=json_set(payload_json,'$.settledOverride',?)${thread ? " WHERE thread_id=?" : ""}`, + ).run(...(thread ? [override, thread] : [override])); +} + +/** Archives (or with null restores) a T3 thread. Without `thread`, every thread. */ +function archiveT3(db: Database, archived: string | null, thread?: string) { + db.query( + `UPDATE orchestration_v2_projection_threads SET archived_at=?${thread ? " WHERE thread_id=?" : ""}`, + ).run(...(thread ? [archived, thread] : [archived])); +} + test("first read-only Codex tool registers a T3 mailbox, the next scan makes it eligible for automatic wake, and later registration retains that", async () => { const f = fixture(); const repo = join(f.root, "repo"); @@ -112,10 +119,10 @@ test("first read-only Codex tool registers a T3 mailbox, the next scan makes it expect(automaticT3Threads(f.registry, join(f.root, ".t3")).size).toBe(0); const source = join(import.meta.dir, "../src/wake-auto-t3.ts"); const code = `import {enrollT3Threads} from ${JSON.stringify(source)}; enrollT3Threads(${JSON.stringify(f.registry)},${JSON.stringify(join(f.root, ".t3"))},"2026-09-29T00:00:00Z",${JSON.stringify(env.SWARMAIL_URL)});`; - f.db.query("UPDATE projection_threads SET settled_override='settled'").run(); + settleT3(f.db, "settled"); expect(await Bun.spawn([process.execPath, "-e", code]).exited).toBe(0); expect(JSON.parse(readFileSync(f.state, "utf8")).wakeThread).toBeUndefined(); - f.db.query("UPDATE projection_threads SET settled_override=NULL").run(); + settleT3(f.db, null); expect(await Bun.spawn([process.execPath, "-e", code]).exited).toBe(0); } expect(JSON.parse(readFileSync(f.state, "utf8")).wakeThread).toBe("thread-1"); @@ -127,20 +134,15 @@ test("first read-only Codex tool registers a T3 mailbox, the next scan makes it } }); -test("after the V2 cutover, enrollment and selection read V2's Codex and Cursor threads, not the frozen V1 copy", async () => { +test("enrollment and selection read only running, active Codex and Cursor threads", async () => { const f = fixture(); const repo = join(f.root, "repo"); Bun.spawnSync(["git", "init", "-q", repo]); f.db.query("INSERT INTO projection_projects VALUES ('project-1',?)").run(repo); - // V2 starts from a copy of V1's database, so its stale running row for native-1 comes along. - f.db.query("UPDATE provider_session_runtime SET last_seen_at='2026-10-03T02:00:00Z'").run(); - const v2 = join(f.root, ".t3/userdata/statev2.sqlite"); - f.db.exec(`VACUUM INTO '${v2}'`); - const db = new Database(v2); - createT3V2Tables(db); - // Imported V1 threads have no provider thread until their next turn. + const db = f.db; + // Imported threads have no provider thread until their next turn. db.query( - "INSERT INTO orchestration_v2_projection_threads VALUES ('thread-1','project-1','t','codex','full-access','default',NULL,'x','x',NULL,NULL,'{}','codex')", + "INSERT INTO orchestration_v2_projection_threads VALUES ('thread-0','project-1','t','codex','full-access','default',NULL,'x','x',NULL,NULL,'{}','codex')", ).run(); addT3V2Thread(db, { threadId: "thread-2", nativeId: "native-2", status: "active" }); addT3V2Thread(db, { @@ -163,7 +165,6 @@ test("after the V2 cutover, enrollment and selection read V2's Codex and Cursor "UPDATE orchestration_v2_projection_threads SET payload_json=json_set(payload_json,?,?) WHERE thread_id=?", ).run(`$.${field}`, value, threadId); } - db.close(); const calls: any[] = []; const server = Bun.serve({ hostname: "127.0.0.1", @@ -201,39 +202,26 @@ test("after the V2 cutover, enrollment and selection read V2's Codex and Cursor ]); expect(existsSync(join(f.registry, "native-8.json"))).toBe(false); expect(existsSync(join(f.registry, "agent-9.json"))).toBe(false); - const current = new Database(v2); - try { - current - .query( - "UPDATE orchestration_v2_projection_threads SET payload_json=json_set(payload_json,'$.settledOverride','settled') WHERE thread_id='thread-2'", - ) - .run(); - current - .query( - "UPDATE orchestration_v2_projection_threads SET payload_json=json_set(payload_json,'$.settledAt','settled') WHERE thread_id='thread-7'", - ) - .run(); - expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))]).toEqual([]); - current - .query( - "UPDATE orchestration_v2_projection_threads SET payload_json=json_set(payload_json,'$.settledOverride','active','$.settledAt',NULL)", - ) - .run(); - expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))].sort()).toEqual([ - "thread-2", - "thread-7", - ]); - const resumed = Bun.spawn([process.execPath, "-e", code], { stdout: "pipe", stderr: "pipe" }); - expect(await resumed.exited, await new Response(resumed.stderr).text()).toBe(0); - expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))].sort()).toEqual([ - "thread-2", - "thread-7", - "thread-8", - "thread-9", - ]); - } finally { - current.close(); - } + settleT3(db, "settled", "thread-2"); + db.query( + "UPDATE orchestration_v2_projection_threads SET payload_json=json_set(payload_json,'$.settledAt','settled') WHERE thread_id='thread-7'", + ).run(); + expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))]).toEqual([]); + db.query( + "UPDATE orchestration_v2_projection_threads SET payload_json=json_set(payload_json,'$.settledOverride','active','$.settledAt',NULL)", + ).run(); + expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))].sort()).toEqual([ + "thread-2", + "thread-7", + ]); + const resumed = Bun.spawn([process.execPath, "-e", code], { stdout: "pipe", stderr: "pipe" }); + expect(await resumed.exited, await new Response(resumed.stderr).text()).toBe(0); + expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))].sort()).toEqual([ + "thread-2", + "thread-7", + "thread-8", + "thread-9", + ]); } finally { await server.stop(true); f.close(); @@ -287,13 +275,13 @@ test("thread selection survives registration, end, resume, archive and provider writeFileSync(f.state, JSON.stringify(state)); // Duplicate state must not spawn a second bridge; an old provider cannot claim its replacement. writeFileSync(join(f.registry, "native-old.json"), JSON.stringify(state)); - f.db.query("UPDATE projection_threads SET archived_at=?").run(archived ? "archived" : null); - f.db - .query("UPDATE projection_threads SET settled_at=?,settled_override=?") - .run(settled ? "settled" : null, settled ? "settled" : null); + archiveT3(f.db, archived ? "archived" : null); + settleT3(f.db, settled ? "settled" : null); f.db - .query("UPDATE provider_session_runtime SET resume_cursor_json=?") - .run(JSON.stringify({ threadId: replaced ? "native-new" : "native-1" })); + .query( + "UPDATE orchestration_v2_projection_provider_threads SET payload_json=json_set(payload_json,'$.nativeThreadRef.nativeId',?)", + ) + .run(replaced ? "native-new" : "native-1"); const selected = [...automaticT3Threads(f.registry, join(f.root, ".t3"))]; expect(selected, sequence.join(" → ")).toEqual( state.wakeThread && !state.ended && !archived && !replaced && !settled ? ["thread-1"] : [], @@ -309,7 +297,11 @@ test("runtime enrollment registers a fresh read-only T3 session without native h const repo = join(f.root, "repo"); Bun.spawnSync(["git", "init", "-q", repo]); f.db.query("INSERT INTO projection_projects VALUES ('project-1',?)").run(repo); - f.db.query("UPDATE provider_session_runtime SET last_seen_at='2026-09-30T22:00:00Z'").run(); + f.db + .query( + "UPDATE orchestration_v2_projection_provider_threads SET updated_at='2026-09-30T22:00:00Z'", + ) + .run(); const calls: any[] = []; const server = Bun.serve({ hostname: "127.0.0.1", @@ -355,7 +347,11 @@ test("enrollment skips a running T3 row last seen before the registration window const repo = join(f.root, "repo"); Bun.spawnSync(["git", "init", "-q", repo]); f.db.query("INSERT INTO projection_projects VALUES ('project-1',?)").run(repo); - f.db.query("UPDATE provider_session_runtime SET last_seen_at='2026-09-30T22:00:00Z'").run(); + f.db + .query( + "UPDATE orchestration_v2_projection_provider_threads SET updated_at='2026-09-30T22:00:00Z'", + ) + .run(); const calls: any[] = []; const server = Bun.serve({ hostname: "127.0.0.1", @@ -401,7 +397,11 @@ for (const mode of [ const repo = join(f.root, "repo"); Bun.spawnSync(["git", "init", "-q", repo]); f.db.query("INSERT INTO projection_projects VALUES ('project-1',?)").run(repo); - f.db.query("UPDATE provider_session_runtime SET last_seen_at='2026-09-30T22:00:00Z'").run(); + f.db + .query( + "UPDATE orchestration_v2_projection_provider_threads SET updated_at='2026-09-30T22:00:00Z'", + ) + .run(); if (mode !== "end-before-first-registration") { writeFileSync( f.state, @@ -413,7 +413,7 @@ for (const mode of [ ); } if (mode === "stopped-after-end") { - f.db.query("UPDATE provider_session_runtime SET status='stopped'").run(); + f.db.query("UPDATE orchestration_v2_projection_provider_threads SET status='closed'").run(); } let release!: () => void; const held = new Promise((done) => { @@ -524,7 +524,7 @@ test("automatic config rejects unverified rotation before touching credentials", JSON.stringify({ swarmailUrl: "http://127.0.0.1:18765", target: { - type: "t3-v1-steer", + type: "t3-v2-queue", url: "http://127.0.0.1:3775", id: "t3-auto-wake", authorizationFile: header, @@ -582,9 +582,10 @@ test("supervisor state survives a T3 upgrade and credential cutover, but not a n expect(() => open(automaticBinding(configure({}, {})))).not.toThrow(); expect(() => open(automaticBinding(configure({ delivery: "auto" }, {})))).not.toThrow(); expect(() => configure({ delivery: "restart" }, {})).toThrow("explicit protocol"); - expect(() => configure({ type: "t3-v1-steer", delivery: "auto" }, {})).toThrow( - "explicit protocol", + expect(() => configure({ type: "t3-v1-steer" }, {})).toThrow( + "target type t3-v1-steer was removed: current T3 Code runs only its V2 engine, which has no V1 route; use t3-v2-queue", ); + expect(() => configure({ type: "opencode-v2-queue" }, {})).toThrow("explicit protocol"); const moved = configure({}, { baseDir: join(root, "other-t3") }); expect(() => open(automaticBinding(moved))).toThrow("binding differs"); } finally { @@ -603,7 +604,11 @@ test("SessionEnd before the first enrollment lock prevents a stale runtime snaps const actualGit = Bun.which("git")!; Bun.spawnSync([actualGit, "init", "-q", repo]); f.db.query("INSERT INTO projection_projects VALUES ('project-1',?)").run(repo); - f.db.query("UPDATE provider_session_runtime SET last_seen_at='2026-09-30T22:00:00Z'").run(); + f.db + .query( + "UPDATE orchestration_v2_projection_provider_threads SET updated_at='2026-09-30T22:00:00Z'", + ) + .run(); const server = Bun.serve({ hostname: "127.0.0.1", port: 0, @@ -654,7 +659,7 @@ import {writeFileSync,existsSync} from "node:fs";writeFileSync(${JSON.stringify( } const { openRegistry } = await import("../src/registry.ts"); openRegistry(dirname(f.state)).end("native-1", () => {}, new Date("2026-09-30T23:00:00Z")); - f.db.query("UPDATE provider_session_runtime SET status='stopped'").run(); + f.db.query("UPDATE orchestration_v2_projection_provider_threads SET status='closed'").run(); writeFileSync(released, "yes"); expect(await child.exited).toBe(0); expect(JSON.parse(readFileSync(f.state, "utf8")).ended).toBe("2026-09-30T23:00:00.000Z"); @@ -674,7 +679,7 @@ test("explicit thread recovery requires the supervisor lock, preserves other fai const stateRoot = join(f.root, "state"); const config = { swarmailUrl: "http://127.0.0.1:1", - target: { type: "t3-v1-steer", id: "t3-auto-wake" }, + target: { type: "t3-v2-queue", id: "t3-auto-wake" }, rotation: { verifyBackend: true, baseDir: join(f.root, ".t3") }, enrollAfter: "2099-01-01T00:00:00Z", }; @@ -703,8 +708,8 @@ test("explicit thread recovery requires the supervisor lock, preserves other fai journal .query("INSERT INTO state VALUES(1,?,4,?)") .run( - JSON.stringify({ type: "t3-v1-steer", url: "http://127.0.0.1:1", id: "thread-1" }), - JSON.stringify({ eventId: 5, command: { commandId: "original-command" } }), + JSON.stringify({ type: "t3-v2-queue", id: "thread-1" }), + JSON.stringify({ eventId: 5, command: { type: "t3.notice", messageId: "original-message" } }), ); journal.close(); const originalMailbox = readFileSync(mailbox); @@ -778,7 +783,7 @@ for (const timing of [ const stateRoot = join(f.root, "state"); const expiresAt = new Date(Date.now() + 86400000).toISOString(); let fatalCommand: any = null; - let pausedDelivery: Resolve | undefined; + let pausedDelivery: ((receipt: object) => void) | undefined; let pauseDelivery = false; let stopBarrier: Resolve | undefined; const stopGate = join(f.root, "stop-gate"); @@ -787,12 +792,13 @@ for (const timing of [ const commands: any[] = []; const requests = new Set(); let authenticated = true; + let refusedTickets = 0; const clock = join(f.root, "clock"); writeFileSync(clock, String(Date.now())); const targetOptions = { hostname: "127.0.0.1", port: 0, - async fetch(request: Request) { + async fetch(request: Request, server: Bun.Server) { const path = new URL(request.url).pathname; requests.add(new URL(request.url).host + path); if (path === "/api/auth/session") { @@ -806,35 +812,55 @@ for (const timing of [ scopes: ["orchestration:operate"], }); } - if (path === "/api/orchestration/dispatch") { - const command = (await request.json()) as any; - commands.push(command); + if (path === "/ws") { + return server.upgrade(request) ? undefined : new Response(null, { status: 400 }); + } + if (path === "/api/auth/websocket-ticket") { if (!authenticated) { + refusedTickets++; return new Response("private-auth-body", { status: 403 }); } - if (command.threadId === "thread-1") { - fatalCommand = command; - } - if (pauseDelivery && command.threadId === pausedThread) { - return new Promise((resolve) => { - pausedDelivery = resolve; - }); - } - return Response.json(command.threadId === "thread-1" ? {} : { sequence: 1 }); + return Response.json({ ticket: "fixture-ticket" }); } if (path.endsWith("/thread-3")) { return Response.json({ private: "private-thread-body" }); } + const id = path.split("/").at(path.endsWith("/bounded") ? -2 : -1); return Response.json({ - thread: { - id: path.split("/").at(-1), - runtimeMode: "approval-required", - settledAt: null, - settledOverride: null, - interactionMode: "default", + projection: { + thread: { id, settledAt: null, settledOverride: null }, + runs: [], + messages: [], + runtimeRequests: [], }, }); }, + websocket: { + message(ws: Bun.ServerWebSocket, raw: string | Buffer) { + const frame = JSON.parse(String(raw)); + if (frame._tag !== "Request") { + return; + } + const command = frame.payload; + commands.push(command); + const reply = (receipt: object) => + ws.send( + JSON.stringify({ + _tag: "Exit", + requestId: frame.id, + exit: { _tag: "Success", value: receipt }, + }), + ); + if (command.threadId === "thread-1") { + fatalCommand = command; + } + if (pauseDelivery && command.threadId === pausedThread) { + pausedDelivery = reply; + return; + } + reply(command.threadId === "thread-1" ? {} : { sequence: 1 }); + }, + }, }; const target = Bun.serve(targetOptions); let movedTarget: Bun.Server | undefined; @@ -855,6 +881,12 @@ for (const timing of [ if (path === "/wait/status") { return Response.json({ eligible: true }); } + if (path === "/wait/peek") { + const session = new URL(request.url).searchParams.get("session"); + return Response.json({ + mailboxes: [{ recipient: `AutoReceiver-${session}`, project: "/w/auto" }], + }); + } if (new URL(request.url).pathname === "/stop") { return new Promise((resolve) => { stopBarrier = resolve; @@ -888,7 +920,7 @@ for (const timing of [ const config = { swarmailUrl: `http://127.0.0.1:${mail.port}`, target: { - type: "t3-v1-steer", + type: "t3-v2-queue", url: `http://127.0.0.1:${target.port}`, id: "t3-auto-wake", authorizationFile: join(f.root, "authorization"), @@ -942,9 +974,7 @@ for (const timing of [ db.close(); } }; - f.db - .exec(`INSERT INTO projection_threads VALUES ('thread-2',NULL,NULL,'project-1',NULL,NULL,NULL); - INSERT INTO provider_session_runtime VALUES ('thread-2','codex','{"threadId":"native-2"}','2026-09-30','running');`); + addT3V2Thread(f.db, { threadId: "thread-2", nativeId: "native-2", updatedAt: "2026-09-30" }); for (const index of [1, 2]) { writeFileSync( join(f.registry, `native-${index}.json`), @@ -1018,9 +1048,7 @@ for (const timing of [ .query<{ value: string }, []>("SELECT value FROM context WHERE id=1") .get()!.value; try { - f.db.exec( - "UPDATE projection_threads SET archived_at='archived' WHERE thread_id='thread-1'", - ); + archiveT3(f.db, "archived", "thread-1"); await scan(); await until(() => Object.keys(status()?.subscriptions ?? {}).length === 1); await until(() => !waiting.has("thread-1")); // The removed bridge stopped polling. @@ -1033,7 +1061,7 @@ for (const timing of [ observer.close(); } expect(journal("thread-1").acknowledged).toBe(0); - f.db.exec("UPDATE projection_threads SET archived_at=NULL WHERE thread_id='thread-1'"); + archiveT3(f.db, null, "thread-1"); await scan(); await until( () => Object.keys(status()?.subscriptions ?? {}).length === 2 && waiting.has("thread-1"), @@ -1045,11 +1073,7 @@ for (const timing of [ waiting.get(pausedThread)!(offer()); await until(() => pausedDelivery); writeFileSync(stopGate, "1"); - f.db.exec( - ownRemoval - ? "UPDATE projection_threads SET archived_at='archived' WHERE thread_id='thread-1'" - : "UPDATE projection_threads SET archived_at='archived'", - ); + archiveT3(f.db, "archived", ownRemoval ? "thread-1" : undefined); await until(() => scanBarrier); if (shuttingDown) { child.kill("SIGTERM"); @@ -1067,7 +1091,7 @@ for (const timing of [ } pauseDelivery = false; rmSync(stopGate); - pausedDelivery!(Response.json(ownRemoval ? {} : { sequence: 1 })); + pausedDelivery!(ownRemoval ? {} : { sequence: 1 }); pausedDelivery = undefined; if (shuttingDown) { expect(await child.exited).toBe(0); @@ -1078,7 +1102,7 @@ for (const timing of [ } stopBarrier!(new Response()); stopBarrier = undefined; - f.db.exec("UPDATE projection_threads SET archived_at=NULL"); + archiveT3(f.db, null); if (shuttingDown) { child = launch(); errors = new Response(child.stderr).text(); @@ -1103,7 +1127,7 @@ for (const timing of [ expect(saved.acknowledged).toBe(0); expect(JSON.parse(saved.pending)).toMatchObject({ eventId: 1, - command: { commandId: fatalCommand.commandId }, + command: { type: "t3.notice", operation: { commandId: fatalCommand.commandId } }, }); // A restarted bridge publishes its subscription before asynchronous retry acceptance. await until(() => journal("thread-2").acknowledged === healthyEvent - 1); @@ -1119,15 +1143,12 @@ for (const timing of [ const next = [...sequence, event]; const before = f.db .query<{ archived_at: string | null; settled_override: string | null }, []>( - "SELECT archived_at, settled_override FROM projection_threads WHERE thread_id='thread-1'", + "SELECT archived_at, json_extract(payload_json,'$.settledOverride') AS settled_override FROM orchestration_v2_projection_threads WHERE thread_id='thread-1'", ) .get()!; if (event !== "scan") { - f.db - .query( - "UPDATE projection_threads SET archived_at=?,settled_override=? WHERE thread_id='thread-1'", - ) - .run(event === "archive" ? "archived" : null, event === "settle" ? "settled" : null); + archiveT3(f.db, event === "archive" ? "archived" : null, "thread-1"); + settleT3(f.db, event === "settle" ? "settled" : null, "thread-1"); } await scan(); expect(Object.keys(status().subscriptions), next.join(" -> ")).toEqual(["thread-2"]); @@ -1137,11 +1158,8 @@ for (const timing of [ next.join(" -> "), ).toHaveLength(1); await visit(next); - f.db - .query( - "UPDATE projection_threads SET archived_at=?,settled_override=? WHERE thread_id='thread-1'", - ) - .run(before.archived_at, before.settled_override); + archiveT3(f.db, before.archived_at, "thread-1"); + settleT3(f.db, before.settled_override, "thread-1"); await scan(); } }; @@ -1170,7 +1188,7 @@ for (const timing of [ if (!shuttingDown) { expect(await errors).toContain("bridge thread-1 quarantined"); } - f.db.exec("UPDATE projection_threads SET archived_at=NULL WHERE thread_id='thread-1'"); + archiveT3(f.db, null, "thread-1"); child = launch(); errors = new Response(child.stderr).text(); await until(() => scanBarrier); @@ -1178,9 +1196,7 @@ for (const timing of [ expect(journal("thread-1").pending).toBe(originalPending); expect(commands.filter((command) => command.threadId === "thread-1")).toHaveLength(1); // A thread-local protocol/preflight failure does not stop a healthy sibling. - f.db - .exec(`INSERT INTO projection_threads VALUES ('thread-3',NULL,NULL,'project-1',NULL,NULL,NULL); - INSERT INTO provider_session_runtime VALUES ('thread-3','codex','{"threadId":"native-3"}','2026-09-30','running');`); + addT3V2Thread(f.db, { threadId: "thread-3", nativeId: "native-3", updatedAt: "2026-09-30" }); writeFileSync( join(f.registry, "native-3.json"), JSON.stringify({ @@ -1195,9 +1211,7 @@ for (const timing of [ expect(Object.keys(status().subscriptions)).toEqual(["thread-2"]); // Shared authentication failure during the cached minute is global, not thread quarantine. authenticated = false; - f.db - .exec(`INSERT INTO projection_threads VALUES ('thread-4',NULL,NULL,'project-1',NULL,NULL,NULL); - INSERT INTO provider_session_runtime VALUES ('thread-4','codex','{"threadId":"native-4"}','2026-09-30','running');`); + addT3V2Thread(f.db, { threadId: "thread-4", nativeId: "native-4", updatedAt: "2026-09-30" }); writeFileSync( join(f.registry, "native-4.json"), JSON.stringify({ @@ -1226,14 +1240,9 @@ for (const timing of [ child = launch(); errors = new Response(child.stderr).text(); await until(() => scanBarrier && waiting.has("thread-2")); - const previousCommands = commands.filter((command) => command.threadId === "thread-2").length; authenticated = false; waiting.get("thread-2")!(offer(healthyEvent + 1)); - await until( - () => - commands.filter((command) => command.threadId === "thread-2").length === - previousCommands + 1 && stopped("thread-2"), - ); + await until(() => refusedTickets === 1 && stopped("thread-2")); scanBarrier!(new Response()); scanBarrier = undefined; await until(() => child.exitCode !== null || status()?.failures?.["thread-2"]); @@ -1246,7 +1255,7 @@ for (const timing of [ if (child.exitCode === null) { child.kill("SIGTERM"); } - pausedDelivery?.(Response.json({ sequence: 1 })); + pausedDelivery?.({ sequence: 1 }); stopBarrier?.(new Response()); scanBarrier?.(new Response()); await child.exited; @@ -1293,13 +1302,13 @@ test("automatic supervisor waits out a T3 restart and follows its new port", asy scopes: ["orchestration:operate"], }); } + const id = path.split("/").at(path.endsWith("/bounded") ? -2 : -1); return Response.json({ - thread: { - id: path.split("/").at(-1), - runtimeMode: "approval-required", - settledAt: null, - settledOverride: null, - interactionMode: "default", + projection: { + thread: { id, settledAt: null, settledOverride: null }, + runs: [], + messages: [], + runtimeRequests: [], }, }); }, @@ -1333,7 +1342,7 @@ test("automatic supervisor waits out a T3 restart and follows its new port", asy const baseDir = join(f.root, ".t3"); const config = { swarmailUrl: `http://127.0.0.1:${mail.port}`, - target: { type: "t3-v1-steer", id: "t3-auto-wake", authorizationFile: join(f.root, "auth") }, + target: { type: "t3-v2-queue", id: "t3-auto-wake", authorizationFile: join(f.root, "auth") }, rotation: { verifyBackend: true, baseDir, @@ -1453,35 +1462,6 @@ test("automatic supervisor waits out a T3 restart and follows its new port", asy ); }, 30000); -test("automatic wake matches the Codex cursor identity, excluding metadata and invalid JSON", () => { - const f = fixture(); - try { - writeFileSync( - f.state, - JSON.stringify({ - name: "BlueLake", - projects: ["/repo"], - wakeThread: "thread-1", - tags: { "/repo": "[t3:thread-1 codex:native-1 cwd:/repo]" }, - }), - ); - for (const cursor of [ - JSON.stringify({ threadId: "native-other", note: "native-1" }), - JSON.stringify({ "native-1": "native-other" }), - '{"threadId":"native-1", broken}', - ]) { - f.db.query("UPDATE provider_session_runtime SET resume_cursor_json=?").run(cursor); - expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))]).toEqual([]); - } - f.db - .query("UPDATE provider_session_runtime SET resume_cursor_json=?") - .run(JSON.stringify({ threadId: "native-1", note: "native-other" })); - expect([...automaticT3Threads(f.registry, join(f.root, ".t3"))]).toEqual(["thread-1"]); - } finally { - f.close(); - } -}); - test("automatic scans repeat only after a T3 commit, a registry write, a reopen or the rescan interval", () => { const f = fixture(); const t3 = heldT3State(join(f.root, ".t3")); @@ -1490,7 +1470,9 @@ test("automatic scans repeat only after a T3 commit, a registry write, a reopen const due = rescanDue(t3, f.registry, () => now); expect(due()).toBe(true); expect(due()).toBe(false); - f.db.query("UPDATE provider_session_runtime SET last_seen_at='2026-10-01'").run(); + f.db + .query("UPDATE orchestration_v2_projection_provider_threads SET updated_at='2026-10-01'") + .run(); expect(due()).toBe(true); expect(due()).toBe(false); Bun.sleepSync(25); // Past the file system's timestamp granularity. @@ -1574,7 +1556,7 @@ for (const initial of ["unavailable", "malformed", "missing"]) { JSON.stringify({ swarmailUrl: `http://127.0.0.1:${proxy.port}`, target: { - type: "t3-v1-steer", + type: "t3-v2-queue", url: `http://127.0.0.1:${target.port}`, authorizationFile: header, }, @@ -1657,8 +1639,7 @@ for (const initial of ["unavailable", "malformed", "missing"]) { test("scan orderings reconcile verified closure and recovery while the T3 backend is unavailable", async () => { const f = fixture(); const sourcePath = join(f.root, ".t3/userdata/statev2.sqlite"); - const source = new Database(sourcePath); - createT3V2Tables(source); + const source = f.db; addT3V2Thread(source, { threadId: "thread-one", nativeId: "native-one" }); source.exec(`CREATE TABLE orchestration_v2_projection_metadata(projection_name TEXT PRIMARY KEY,schema_version INTEGER,last_sequence INTEGER); INSERT INTO orchestration_v2_projection_metadata VALUES ('thread-projections',2,1); @@ -1834,7 +1815,6 @@ test("scan orderings reconcile verified closure and recovery while the T3 backen await proxy.stop(true); await core.server.stop(true); core.db.close(); - source.close(); f.close(); } expect(await errors).toContain( diff --git a/test/wake-backend.test.ts b/test/wake-backend.test.ts index a27f784..11a25bf 100644 --- a/test/wake-backend.test.ts +++ b/test/wake-backend.test.ts @@ -12,7 +12,7 @@ test.skipIf(process.platform !== "linux")( async () => { const root = mkdtempSync(join(scratch, "case-")); mkdirSync(join(root, "userdata")); - const path = join(root, "userdata/state.sqlite"); + const path = join(root, "userdata/statev2.sqlite"); const database = new Database(path, { create: true }); const server = Bun.serve({ hostname: "127.0.0.1", @@ -89,16 +89,6 @@ test.skipIf(process.platform !== "linux")( expect(verifyT3Backend(config)).rejects.toThrow("backend identity"); renameSync(path + ".old", path); expect(verifyT3Backend(config)).resolves.toBe(listener); - // Once V2 has created its database, the listener must hold that one, not V1's frozen file. - const v2Path = join(root, "userdata/statev2.sqlite"); - new Database(v2Path, { create: true }).close(); - expect(verifyT3Backend(config)).rejects.toThrow("backend identity"); - const v2 = new Database(v2Path); - try { - expect(verifyT3Backend(config)).resolves.toBe(listener); - } finally { - v2.close(); - } await server.stop(true); expect(verifyT3Backend(config)).rejects.toThrow("unavailable; retrying"); expect(discover()).rejects.toThrow("unavailable; retrying"); diff --git a/test/wake-credentials.test.ts b/test/wake-credentials.test.ts index 3226d46..95d809a 100644 --- a/test/wake-credentials.test.ts +++ b/test/wake-credentials.test.ts @@ -108,7 +108,7 @@ else if(args[2]==='revoke'){ }; } = { target: { - type: "t3-v1-steer", + type: "t3-v2-queue", id: "test-thread", url: `http://127.0.0.1:${server.port}`, authorizationFile: join(root, "authorization"), @@ -291,6 +291,12 @@ else if(args[2]==='revoke'){ f.configure(); expect((await f.run()).err).toContain("binding differs"); f.config.target.id = "test-thread"; + f.config.target.type = "t3-v1-steer"; + f.configure(); + expect((await f.run()).err).toContain( + "target type t3-v1-steer was removed: current T3 Code runs only its V2 engine, which has no V1 route; use t3-v2-queue", + ); + f.config.target.type = "t3-v2-queue"; f.configure(); chmodSync(f.config.target.authorizationFile, 0o644); expect((await f.run()).err).toContain("private, user-owned"); @@ -570,7 +576,7 @@ else if(args[2]==='revoke'){ const baseDir = join(root, ".t3"); mkdirSync(join(baseDir, "userdata"), { recursive: true }); // verifyT3Backend requires the listener to hold T3's database open; this process is the listener. - const database = new Database(join(baseDir, "userdata/state.sqlite")); + const database = new Database(join(baseDir, "userdata/statev2.sqlite")); const requests: string[] = []; const server = Bun.serve({ hostname: "127.0.0.1", @@ -588,7 +594,7 @@ else if(args[2]==='revoke'){ servers.push(server); const expiresAt = new Date(Date.now() + 86400000).toISOString(); const config = { - target: { type: "t3-v1-steer", id: "minute", authorizationFile: join(root, "auth/header") }, + target: { type: "t3-v2-queue", id: "minute", authorizationFile: join(root, "auth/header") }, rotation: { executable: process.execPath, baseDir, diff --git a/test/wake-t3-backend.test.ts b/test/wake-t3-backend.test.ts index b0a3d75..9855393 100644 --- a/test/wake-t3-backend.test.ts +++ b/test/wake-t3-backend.test.ts @@ -29,7 +29,7 @@ function fixture() { const root = join(scratch, crypto.randomUUID()); const baseDir = join(root, "backend"); mkdirSync(join(baseDir, "userdata"), { recursive: true }); - const database = new Database(join(baseDir, "userdata/state.sqlite")); + const database = new Database(join(baseDir, "userdata/statev2.sqlite")); const server = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: () => new Response("unused") }); cleanup.push( () => database.close(), @@ -196,11 +196,11 @@ linuxTest( f.publish(process.pid + 1000000); expect(await poll()).toBeUndefined(); expect(fetch).toHaveBeenCalledTimes(1); - // A different database path is a fatal identity mismatch. + // A database file the listener does not hold is a fatal identity mismatch. f.publish(); const dbPath = join(f.baseDir, "userdata/statev2.sqlite"); - const other = new Database(dbPath); - other.close(); + renameSync(dbPath, dbPath + ".old"); + new Database(dbPath).close(); expect(poll()).rejects.toBeInstanceOf(BridgeError); expect(fetch).toHaveBeenCalledTimes(1); expect(f.snapshot()).toEqual(before); From 040844397c99ac838123c4d81adb2ba455fdadef Mon Sep 17 00:00:00 2001 From: bompus Date: Mon, 19 Oct 2026 04:17:00 -0600 Subject: [PATCH 2/2] docs: say discovery needs exactly one matching T3 thread; explain switching a t3-v1-steer bridge's saved state --- CHANGELOG.md | 5 ++++- docs/usage.md | 6 +++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bab405f..bd8919b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,10 @@ Notable changes to Swarmail. Versions follow [semantic versioning](https://semve engine: it keeps its data in `statev2.sqlite` and no longer has the HTTP dispatch route or turn-start command those bridges sent, so a `t3-v1-steer` bridge failed its thread check and delivered nothing. A `wake-bridge`, `wake-auto-t3` or credential rotation configuration that still names `t3-v1-steer` now fails at startup with - `target type t3-v1-steer was removed`; change its `target.type` to `t3-v2-queue`. The register hook, `who`, agent + `target type t3-v1-steer was removed`; change its `target.type` to `t3-v2-queue`. The old bridge's saved state is + bound to the `t3-v1-steer` destination, so the switched bridge refuses it with `saved mailbox binding differs`: stop + the old bridge, reconcile or discard its pending delivery as `docs/wake-bridges.md` describes for bridge state, then + remove that journal before starting the `t3-v2-queue` bridge. The register hook, `who`, agent locations, automatic wake and `pr-sessions` read only `statev2.sqlite`. While it is missing, or T3 Code is still creating it, they find no T3 threads, and `pr-sessions` exits with an error. diff --git a/docs/usage.md b/docs/usage.md index cce0677..042c6d1 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -370,9 +370,9 @@ The CLI uses one recognized host session variable, or one registration tied to the current host process's PID and start time. Conflicting variables or several possible sessions fail instead of choosing one. T3 thread identity also matches registrations made by an earlier provider in that thread. Looking the thread up -in T3 Code's database requires one matching provider and native ID. If -discovery fails, -use the explicit mailbox interface below. Keep registration tags current. +in T3 Code's database needs exactly one T3 thread whose provider and native +session ID match the current session; no match, or more than one, makes +discovery fail. If discovery fails, use the explicit mailbox interface below. Keep registration tags current. `--limit N` sets the page size, capped at 1000, not a total limit. The drain stops on a page smaller than that effective limit. `--json` prints one JSON