From 76ee87539e7627637ac5ec7b103aece57b200c94 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 25 Jul 2026 19:05:22 -0700 Subject: [PATCH] feat: detect external human turns in adopted Pi and OpenCode sessions (#113957) * feat(plugins): support additional upstream session kinds * feat(acpx): monitor upstream Pi sessions * feat(opencode): monitor upstream OpenCode sessions * fix(opencode): ignore hidden user text activity * refactor(plugins): share upstream echo filtering * fix(opencode): preserve marker tuple narrowing * refactor(opencode): detect upstream turns via event_sequence cursor Replace timestamp/grace-window completeness heuristics with OpenCode's own per-session event_sequence.seq cursor, which advances inside the same transaction as the message/part projections. Detects change instead of inferring completion, so staged part writes can no longer drop a turn. Handles cursor regression from OpenCode migrations that clear event_sequence, suppresses compaction replay and summary re-publish, and restricts human-turn classification to visible non-synthetic text. * docs(plugins): record why the adoption coordinator requires complete * fix(opencode): align cursor proof with strict types * fix(acpx): keep Pi JSON parser private --- docs/concepts/session-state.md | 7 +- .../pi-session-catalog-continuation.test.ts | 128 ++++ .../acpx/src/pi-session-catalog-plugin.ts | 8 +- .../acpx/src/pi-session-catalog.test.ts | 46 -- extensions/acpx/src/pi-session-catalog.ts | 2 +- extensions/acpx/src/pi-session-store.ts | 27 +- .../src/pi-session-upstream-activity.test.ts | 155 +++++ .../acpx/src/pi-session-upstream-activity.ts | 189 +++++ .../anthropic/session-upstream-activity.ts | 10 +- extensions/opencode/session-catalog-plugin.ts | 14 +- extensions/opencode/session-catalog.test.ts | 19 +- extensions/opencode/session-catalog.ts | 16 +- .../session-upstream-activity.test.ts | 653 ++++++++++++++++++ .../opencode/session-upstream-activity.ts | 429 ++++++++++++ src/plugin-sdk/session-catalog.ts | 2 + src/plugins/session-catalog.ts | 47 +- 16 files changed, 1671 insertions(+), 81 deletions(-) create mode 100644 extensions/acpx/src/pi-session-catalog-continuation.test.ts create mode 100644 extensions/acpx/src/pi-session-upstream-activity.test.ts create mode 100644 extensions/acpx/src/pi-session-upstream-activity.ts create mode 100644 extensions/opencode/session-upstream-activity.test.ts create mode 100644 extensions/opencode/session-upstream-activity.ts diff --git a/docs/concepts/session-state.md b/docs/concepts/session-state.md index a6fda20aeeec..770b7fd9c0ef 100644 --- a/docs/concepts/session-state.md +++ b/docs/concepts/session-state.md @@ -48,7 +48,9 @@ Watcher identity must be an agent-qualified session key. Under `session.scope="g Watches clean themselves up: cursor rows expire with signal-log retention, are removed when the watcher session resets, and are deleted with either session. There is no unwatch verb in v1. -Watched sessions adopted from a session catalog are checked for direct upstream human activity on a fixed cadence. Detected activity enters the same signal log and watcher flow as other direct human turns. +Watched Claude, Codex, OpenCode, and Pi sessions adopted from a session catalog are checked for direct upstream human activity on a fixed cadence. Pi monitoring starts after the session is in its append-only v3 format. Detected activity enters the same signal log and watcher flow as other direct human turns. + +OpenCode detection is deliberately conservative. OpenCode's v1 tables do not preserve message provenance, so reporting ambiguous rows would create false alarms; per-message provenance exists only in its v2 schema. OpenCode therefore does not report image-only turns, `@file`-mention-only turns, slash commands routed to a subagent, or turns from ACP clients that annotate content with an audience (mapped by OpenCode to `synthetic` or `ignored`). It also suppresses text matching any of the preceding 50 user messages to catch compaction replay, which means a human deliberately repeating the same text within that window can be missed. If an adopted session's upstream source is deleted externally, three consecutive missing checks (about three monitor ticks) produce one `upstream_missing` signal for its watchers and remove the upstream link. Continuing the catalog session again creates a fresh link. @@ -105,6 +107,9 @@ Current limits: - Cancelled-outcome payload detail is currently produced by ACP child runs; native sub-agent cancellations surface as generic failures. - Upstream self-echo detection compares normalized user text. An external prompt matching one of the session's 10 most recent OpenClaw-side user messages is treated as self-echo. - A single local Claude JSONL row larger than the 1 MiB per-cadence scan cap blocks that session's cursor in v1; unclassified bytes are never skipped. +- A single Pi JSONL row larger than the 1 MiB per-cadence scan cap blocks that session's cursor in v1; unclassified bytes are never skipped. +- Legacy Pi sessions are adopted without an upstream link. Resume once to migrate the file to v3, then continue it from the catalog again to start monitoring. +- OpenCode checks issue one batched database query per cadence. A session export runs only when that query shows its durable event sequence advanced. - Paired-node Claude checks classify the latest 50 transcript items per cadence. Larger bursts can fall outside the v1 scan window. - Paired-node Claude history reads do not expose a definitive thread-not-found result, so remote Claude deletions are not classified as `upstream_missing` in v1. - Catalog sessions that have not been adopted remain outside the awareness layer in v1. diff --git a/extensions/acpx/src/pi-session-catalog-continuation.test.ts b/extensions/acpx/src/pi-session-catalog-continuation.test.ts new file mode 100644 index 000000000000..ec4ee138400e --- /dev/null +++ b/extensions/acpx/src/pi-session-catalog-continuation.test.ts @@ -0,0 +1,128 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +type ResolveAcpSessionAvailability = + (typeof import("openclaw/plugin-sdk/acp-runtime"))["resolveAcpSessionAvailability"]; + +const acpRuntimeMocks = vi.hoisted(() => ({ + resolveAcpSessionAvailability: vi.fn(() => ({ available: true })), +})); + +vi.mock("openclaw/plugin-sdk/acp-runtime", async (importOriginal) => ({ + ...(await importOriginal()), + resolveAcpSessionAvailability: acpRuntimeMocks.resolveAcpSessionAvailability, +})); + +import { + capturePiContinuationCatalog, + createPiStoreFixture, + installFakePiFixture, +} from "./pi-session-catalog.test-support.js"; + +const temporaryDirectories: string[] = []; +const originalSessionDir = process.env.PI_CODING_AGENT_SESSION_DIR; +const originalAgentDir = process.env.PI_CODING_AGENT_DIR; +const originalPath = process.env.PATH; + +afterEach(async () => { + acpRuntimeMocks.resolveAcpSessionAvailability.mockReset().mockReturnValue({ available: true }); + process.env.PATH = originalPath; + if (originalSessionDir === undefined) { + delete process.env.PI_CODING_AGENT_SESSION_DIR; + } else { + process.env.PI_CODING_AGENT_SESSION_DIR = originalSessionDir; + } + if (originalAgentDir === undefined) { + delete process.env.PI_CODING_AGENT_DIR; + } else { + process.env.PI_CODING_AGENT_DIR = originalAgentDir; + } + await Promise.all( + temporaryDirectories.splice(0).map(async (directory) => { + await fs.rm(directory, { recursive: true, force: true }); + }), + ); +}); + +describe("Pi session catalog continuation", () => { + it("adopts once with the native ACP binding and an exact file baseline", async () => { + const sessionDirectory = await createPiStoreFixture( + temporaryDirectories, + "hi", + "Pi catalog session", + { command: "pwd" }, + true, + ); + await installFakePiFixture(temporaryDirectories, originalPath); + const { createSessionEntry, provider } = capturePiContinuationCatalog(); + + const [first, concurrent] = await Promise.all([ + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + ]); + const second = await provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }); + const sessionFile = await fs.realpath(path.join(sessionDirectory, "session.jsonl")); + const sessionStats = await fs.stat(sessionFile); + + expect(first).toEqual(concurrent); + expect(second).toEqual(first); + expect(first.upstream).toEqual({ + kind: "pi-cli", + ref: { filePath: sessionFile }, + marker: expect.objectContaining({ offset: sessionStats.size }), + }); + expect(createSessionEntry).toHaveBeenCalledTimes(1); + expect(createSessionEntry).toHaveBeenCalledWith( + expect.objectContaining({ + label: "Pi catalog session", + spawnedCwd: "/workspace", + initialEntry: { + acpBackendId: "acpx", + acpSessionBinding: { acpAgentId: "pi", agentSessionId: "pi-session" }, + pluginExtensions: { acpx: { piSessionCatalog: { sourceThreadId: "pi-session" } } }, + }, + }), + ); + }); + + it("rejects paired-node and unknown session continuation", async () => { + await createPiStoreFixture( + temporaryDirectories, + "hi", + "Pi catalog session", + { command: "pwd" }, + true, + ); + await installFakePiFixture(temporaryDirectories, originalPath); + const { createSessionEntry, provider } = capturePiContinuationCatalog(); + + await expect( + provider.continueSession!({ hostId: "node:remote", threadId: "pi-session" }), + ).rejects.toThrow("paired-node Pi session rows are view-only"); + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "missing" }), + ).rejects.toThrow("Pi session is unavailable"); + expect(createSessionEntry).not.toHaveBeenCalled(); + }); + + it("keeps legacy-session adoption successful when a safe baseline is unavailable", async () => { + const sessionDirectory = await createPiStoreFixture( + temporaryDirectories, + "hi", + "Pi catalog session", + { command: "pwd" }, + true, + ); + const sessionFile = path.join(sessionDirectory, "session.jsonl"); + const content = await fs.readFile(sessionFile, "utf8"); + await fs.writeFile(sessionFile, content.replace('"version":3', '"version":2')); + await installFakePiFixture(temporaryDirectories, originalPath); + const { createSessionEntry, provider } = capturePiContinuationCatalog(); + + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + ).resolves.toEqual({ sessionKey: expect.any(String) }); + expect(createSessionEntry).toHaveBeenCalledTimes(1); + }); +}); diff --git a/extensions/acpx/src/pi-session-catalog-plugin.ts b/extensions/acpx/src/pi-session-catalog-plugin.ts index 5e503c3cd455..0bd1da5ef801 100644 --- a/extensions/acpx/src/pi-session-catalog-plugin.ts +++ b/extensions/acpx/src/pi-session-catalog-plugin.ts @@ -35,6 +35,7 @@ import { type PiSessionPage, } from "./pi-session-catalog.js"; import { piSessionStoreAvailable } from "./pi-session-paths.js"; +import { checkPiUpstreamActivity, linkContinuedPiSession } from "./pi-session-upstream-activity.js"; const PI_SESSIONS_LIST_COMMAND = "acpx.pi.sessions.list.v1"; const PI_SESSION_READ_COMMAND = "acpx.pi.sessions.read.v1"; @@ -60,7 +61,8 @@ const PI_ADOPTED_SESSION_KEY_PREFIX = "plugin:acpx:catalog-adopt:pi:"; class PiCatalogParamsError extends Error {} -const continueAdoption = createSessionCatalogAdoptionCoordinator(); +const continueAdoption = + createSessionCatalogAdoptionCoordinator>>(); function validatePiThreadId(value: unknown): string { if (typeof value !== "string" || !SESSION_ID_PATTERN.test(value)) { @@ -446,7 +448,7 @@ async function continuePiSession( api: OpenClawPluginApi, hostId: string, threadId: string, -): Promise<{ sessionKey: string }> { +): Promise>> { if (hostId.startsWith("node:")) { throw new PiCatalogParamsError("paired-node Pi session rows are view-only"); } @@ -500,6 +502,7 @@ async function continuePiSession( }); return { sessionKey: created.key }; }, + complete: async (continued) => await linkContinuedPiSession(continued.sessionKey, threadId), }); } @@ -635,6 +638,7 @@ export function registerPiSessionCatalog(api: OpenClawPluginApi): void { read: async (request) => await readPiTranscript(api.runtime, request), continueSession: async (request) => await continuePiSession(api, request.hostId, request.threadId), + checkUpstreamActivity: checkPiUpstreamActivity, openTerminal: async (request) => await openPiTerminal({ runtime: api.runtime, ...request }), }); for (const command of createPiSessionNodeHostCommands()) { diff --git a/extensions/acpx/src/pi-session-catalog.test.ts b/extensions/acpx/src/pi-session-catalog.test.ts index f61194fad3ce..0e3c86192c8a 100644 --- a/extensions/acpx/src/pi-session-catalog.test.ts +++ b/extensions/acpx/src/pi-session-catalog.test.ts @@ -679,52 +679,6 @@ describe("Pi session catalog", () => { }, ); - it("adopts local Pi sessions once with the native ACP resume binding", async () => { - await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); - await installFakePi(); - const { createSessionEntry, provider } = capturePiContinuationCatalog(); - - const [first, concurrent] = await Promise.all([ - provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), - provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), - ]); - const second = await provider.continueSession!({ - hostId: "gateway", - threadId: "pi-session", - }); - - expect(first).toEqual(concurrent); - expect(second).toEqual(first); - expect(createSessionEntry).toHaveBeenCalledTimes(1); - expect(createSessionEntry).toHaveBeenCalledWith( - expect.objectContaining({ - label: "Pi catalog session", - spawnedCwd: "/workspace", - initialEntry: { - acpBackendId: "acpx", - acpSessionBinding: { acpAgentId: "pi", agentSessionId: "pi-session" }, - pluginExtensions: { - acpx: { piSessionCatalog: { sourceThreadId: "pi-session" } }, - }, - }, - }), - ); - }); - - it("rejects paired-node and unknown Pi session continuation", async () => { - await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); - await installFakePi(); - const { createSessionEntry, provider } = capturePiContinuationCatalog(); - - await expect( - provider.continueSession!({ hostId: "node:remote", threadId: "pi-session" }), - ).rejects.toThrow("paired-node Pi session rows are view-only"); - await expect( - provider.continueSession!({ hostId: "gateway", threadId: "missing" }), - ).rejects.toThrow("Pi session is unavailable"); - expect(createSessionEntry).not.toHaveBeenCalled(); - }); - it("hides and rejects Continue when ACP cannot resume Pi", async () => { await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); await installFakePi(); diff --git a/extensions/acpx/src/pi-session-catalog.ts b/extensions/acpx/src/pi-session-catalog.ts index e8073759b41a..22e0427c76d1 100644 --- a/extensions/acpx/src/pi-session-catalog.ts +++ b/extensions/acpx/src/pi-session-catalog.ts @@ -225,7 +225,7 @@ export async function listLocalPiSessionPage(value?: unknown): Promise session); + const page = summaries.map(({ file: _file, version: _version, ...session }) => session); return { sessions: page, ...(hasMore ? { nextCursor: encodeCursor(offset + page.length) } : {}), diff --git a/extensions/acpx/src/pi-session-store.ts b/extensions/acpx/src/pi-session-store.ts index cbcb317a91a8..53474f461afc 100644 --- a/extensions/acpx/src/pi-session-store.ts +++ b/extensions/acpx/src/pi-session-store.ts @@ -14,7 +14,7 @@ const APPEND_PROOF_EDGE_BYTES = 64 * 1024; const IO_CONCURRENCY = 8; const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; -type PiSessionSummary = SessionCatalogSession & { file: string }; +type PiSessionSummary = SessionCatalogSession & { file: string; version: number }; type PiFileCandidate = { file: string; @@ -374,12 +374,15 @@ async function readPiSessionSummary( processSummaryLine(projectedState, projectedState.pending); } const { header, name, firstMessage } = projectedState; + const version = + header?.type === "session" && typeof header.version === "number" ? header.version : 1; const threadId = header?.type === "session" ? optionalString(header.id, 256) : undefined; if (header && threadId && SESSION_ID_PATTERN.test(threadId)) { const cwd = optionalString(header.cwd, 4_096); const createdAt = timestampMs(header.timestamp); summary = { file: candidate.file, + version, threadId, ...(name || firstMessage ? { name: name ?? firstMessage } : {}), ...(cwd ? { cwd } : {}), @@ -473,6 +476,28 @@ async function findPiSummary( return undefined; } +export async function readPiSessionFileBaseline( + threadId: string, + env: NodeJS.ProcessEnv, +): Promise< + | { + filePath: string; + offset: number; + } + | undefined +> { + const summary = await findPiSummary(threadId, env); + if (!summary?.canContinue || summary.version < 3) { + return undefined; + } + try { + const stats = await fs.stat(summary.file); + return stats.isFile() ? { filePath: summary.file, offset: stats.size } : undefined; + } catch { + return undefined; + } +} + export async function readPiSessionById( threadId: string, env: NodeJS.ProcessEnv, diff --git a/extensions/acpx/src/pi-session-upstream-activity.test.ts b/extensions/acpx/src/pi-session-upstream-activity.test.ts new file mode 100644 index 000000000000..085d62ad2c8d --- /dev/null +++ b/extensions/acpx/src/pi-session-upstream-activity.test.ts @@ -0,0 +1,155 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { createPiStoreFixture } from "./pi-session-catalog.test-support.js"; +import { checkPiUpstreamActivity, linkContinuedPiSession } from "./pi-session-upstream-activity.js"; + +const temporaryDirectories: string[] = []; +const originalSessionDir = process.env.PI_CODING_AGENT_SESSION_DIR; +const originalAgentDir = process.env.PI_CODING_AGENT_DIR; + +afterEach(async () => { + if (originalSessionDir === undefined) { + delete process.env.PI_CODING_AGENT_SESSION_DIR; + } else { + process.env.PI_CODING_AGENT_SESSION_DIR = originalSessionDir; + } + if (originalAgentDir === undefined) { + delete process.env.PI_CODING_AGENT_DIR; + } else { + process.env.PI_CODING_AGENT_DIR = originalAgentDir; + } + await Promise.all( + temporaryDirectories.splice(0).map(async (directory) => { + await fs.rm(directory, { recursive: true, force: true }); + }), + ); +}); + +describe("Pi session upstream activity", () => { + it("detects external turns, suppresses own echoes, and confirms deletion", async () => { + const sessionDirectory = await createPiStoreFixture( + temporaryDirectories, + "hi", + "Pi catalog session", + { command: "pwd" }, + true, + ); + const continued = await linkContinuedPiSession("agent:main:pi", "pi-session"); + const file = path.join(sessionDirectory, "session.jsonl"); + const probe = { + sessionKey: continued.sessionKey, + agentId: "main", + threadId: "pi-session", + hostId: "gateway", + upstreamKind: continued.upstream!.kind, + upstreamRef: continued.upstream!.ref, + marker: continued.upstream!.marker, + ownRecentUserTexts: ["sent from OpenClaw"], + }; + + await fs.appendFile( + file, + `${JSON.stringify({ + type: "message", + id: "user-own", + parentId: "info-1", + timestamp: "2026-07-13T10:00:05.000Z", + message: { role: "user", content: "sent from OpenClaw" }, + })}\n`, + ); + const ownEcho = await checkPiUpstreamActivity([probe]); + expect(ownEcho).toEqual([ + expect.objectContaining({ + kind: "activity", + sessionKey: continued.sessionKey, + humanTurns: 0, + }), + ]); + + await fs.appendFile( + file, + `${JSON.stringify({ + type: "message", + id: "user-external", + parentId: "user-own", + timestamp: "2026-07-13T10:00:06.000Z", + message: { role: "user", content: "external Pi turn" }, + })}\n`, + ); + const external = await checkPiUpstreamActivity([ + { ...probe, marker: ownEcho[0]!.kind === "activity" ? ownEcho[0]!.nextMarker : null }, + ]); + expect(external).toEqual([ + expect.objectContaining({ + kind: "activity", + sessionKey: continued.sessionKey, + humanTurns: 1, + }), + ]); + + await fs.rm(file); + await expect(checkPiUpstreamActivity([probe])).resolves.toEqual([ + { kind: "missing", sessionKey: continued.sessionKey }, + ]); + }); + + it("does not classify transient read failures as missing", async () => { + await expect( + checkPiUpstreamActivity([ + { + sessionKey: "agent:main:pi", + agentId: "main", + threadId: "pi-session", + hostId: "gateway", + upstreamKind: "pi-cli", + upstreamRef: { filePath: `/${"x".repeat(10_000)}` }, + marker: { offset: 0 }, + ownRecentUserTexts: [], + }, + ]), + ).resolves.toEqual([]); + }); + + it("stops the cursor before the first unclassifiable complete row", async () => { + const sessionDirectory = await createPiStoreFixture( + temporaryDirectories, + "hi", + "Pi catalog session", + { command: "pwd" }, + true, + ); + const continued = await linkContinuedPiSession("agent:main:pi", "pi-session"); + const file = path.join(sessionDirectory, "session.jsonl"); + const probe = { + sessionKey: continued.sessionKey, + agentId: "main", + threadId: "pi-session", + hostId: "gateway", + upstreamKind: continued.upstream!.kind, + upstreamRef: continued.upstream!.ref, + marker: continued.upstream!.marker, + ownRecentUserTexts: [], + }; + await fs.appendFile( + file, + `${JSON.stringify({ + type: "message", + timestamp: "2026-07-13T10:00:05.000Z", + message: { role: "user", content: "first external turn" }, + })}\n{not-json}\n${JSON.stringify({ + type: "message", + timestamp: "2026-07-13T10:00:06.000Z", + message: { role: "user", content: "must remain behind the bad row" }, + })}\n`, + ); + + const first = await checkPiUpstreamActivity([probe]); + expect(first).toEqual([expect.objectContaining({ kind: "activity", humanTurns: 1 })]); + await expect( + checkPiUpstreamActivity([ + { ...probe, marker: first[0]!.kind === "activity" ? first[0]!.nextMarker : null }, + ]), + ).resolves.toEqual([]); + }); +}); diff --git a/extensions/acpx/src/pi-session-upstream-activity.ts b/extensions/acpx/src/pi-session-upstream-activity.ts new file mode 100644 index 000000000000..eddfb6cef2ed --- /dev/null +++ b/extensions/acpx/src/pi-session-upstream-activity.ts @@ -0,0 +1,189 @@ +import fs from "node:fs/promises"; +import process from "node:process"; +import { + isExternalUserText, + type SessionCatalogContinueProviderResult, + type SessionUpstreamActivity, + type SessionUpstreamProbe, +} from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { readPiSessionFileBaseline } from "./pi-session-store.js"; + +const MAX_PI_UPSTREAM_SCAN_BYTES = 1024 * 1024; + +function parseCompletePiRows(tail: Buffer): { + entries: Record[]; + classifiedBytes: number; +} { + const entries: Record[] = []; + let lineStart = 0; + let classifiedBytes = 0; + for (let index = 0; index < tail.length; index += 1) { + if (tail[index] !== 0x0a) { + continue; + } + const line = tail.subarray(lineStart, index).toString("utf8").trim(); + if (line) { + try { + const value = JSON.parse(line) as unknown; + if (!isRecord(value)) { + break; + } + entries.push(value); + } catch { + break; + } + } + classifiedBytes = index + 1; + lineStart = index + 1; + } + return { entries, classifiedBytes }; +} + +function textFromContent(content: unknown): string | undefined { + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return undefined; + } + const text = content + .flatMap((part) => + isRecord(part) && part.type === "text" && typeof part.text === "string" ? [part.text] : [], + ) + .join("\n"); + return text || undefined; +} + +function timestampMs(value: unknown): number | undefined { + if (typeof value === "number" && Number.isFinite(value)) { + return value; + } + if (typeof value === "string") { + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? undefined : parsed; + } + return undefined; +} + +function readFilePath(probe: SessionUpstreamProbe): string | undefined { + return isRecord(probe.upstreamRef) && typeof probe.upstreamRef.filePath === "string" + ? probe.upstreamRef.filePath + : undefined; +} + +function readMarkerOffset(probe: SessionUpstreamProbe): number | undefined { + return isRecord(probe.marker) && + Number.isSafeInteger(probe.marker.offset) && + Number(probe.marker.offset) >= 0 + ? Number(probe.marker.offset) + : undefined; +} + +export async function linkContinuedPiSession( + sessionKey: string, + threadId: string, +): Promise { + try { + const baseline = await readPiSessionFileBaseline(threadId, process.env); + // Legacy Pi versions rewrite during first resume, so the store helper declines + // those links. Current v3 sessions persist by appending to this exact file. + return baseline + ? { + sessionKey, + upstream: { + kind: "pi-cli", + ref: { filePath: baseline.filePath }, + marker: { offset: baseline.offset }, + }, + } + : { sessionKey }; + } catch { + // Liveness metadata is optional; continuation success must survive baseline failure. + return { sessionKey }; + } +} + +async function checkPiSessionUpstreamActivity( + probe: SessionUpstreamProbe, +): Promise { + if (probe.hostId !== "gateway" || probe.upstreamKind !== "pi-cli") { + return undefined; + } + const filePath = readFilePath(probe); + const markerOffset = readMarkerOffset(probe); + if (!filePath || markerOffset === undefined) { + return undefined; + } + let handle: Awaited>; + try { + handle = await fs.open(filePath, "r"); + } catch (error) { + return isRecord(error) && error.code === "ENOENT" + ? { kind: "missing", sessionKey: probe.sessionKey } + : undefined; + } + try { + const stat = await handle.stat(); + if (!stat.isFile()) { + return { kind: "missing", sessionKey: probe.sessionKey }; + } + if (stat.size <= markerOffset) { + return undefined; + } + const readLength = Math.min(stat.size - markerOffset, MAX_PI_UPSTREAM_SCAN_BYTES); + const buffer = Buffer.allocUnsafe(readLength); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, markerOffset); + const tail = buffer.subarray(0, bytesRead); + const { entries, classifiedBytes } = parseCompletePiRows(tail); + if (classifiedBytes === 0) { + // Never advance past an invalid, partial, or over-cap JSONL row. + return undefined; + } + let humanTurns = 0; + let occurredAt: number | undefined; + for (const entry of entries) { + if (entry.type !== "message" || !isRecord(entry.message) || entry.message.role !== "user") { + continue; + } + const text = textFromContent(entry.message.content); + if (!isExternalUserText(probe, text)) { + continue; + } + humanTurns += 1; + occurredAt = Math.max( + occurredAt ?? 0, + timestampMs(entry.message.timestamp) ?? timestampMs(entry.timestamp) ?? stat.mtimeMs, + ); + } + const nextOffset = markerOffset + classifiedBytes; + return { + kind: "activity", + sessionKey: probe.sessionKey, + humanTurns, + nextMarker: { offset: nextOffset }, + ...(humanTurns > 0 + ? { occurredAt: occurredAt ?? stat.mtimeMs, dedupeId: String(nextOffset) } + : {}), + }; + } finally { + await handle.close(); + } +} + +export async function checkPiUpstreamActivity( + probes: SessionUpstreamProbe[], +): Promise { + const outcomes: SessionUpstreamActivity[] = []; + for (const probe of probes) { + try { + const outcome = await checkPiSessionUpstreamActivity(probe); + if (outcome) { + outcomes.push(outcome); + } + } catch { + // One transient file read must not suppress healthy sessions in the batch. + } + } + return outcomes; +} diff --git a/extensions/anthropic/session-upstream-activity.ts b/extensions/anthropic/session-upstream-activity.ts index c7cf32e1d881..3d8f36e14a68 100644 --- a/extensions/anthropic/session-upstream-activity.ts +++ b/extensions/anthropic/session-upstream-activity.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import { classifyClaudeCliHistoryMessage, classifyClaudeCliHistoryLine, + isExternalUserText, type SessionCatalogContinueProviderResult, type SessionUpstreamActivity, type SessionUpstreamProbe, @@ -104,15 +105,6 @@ function readMarkerOffset(probe: SessionUpstreamProbe): number | undefined { return Number.isSafeInteger(offset) && (offset as number) >= 0 ? (offset as number) : undefined; } -function normalizeUserText(text: string): string { - return text.trim().replace(/\s+/g, " "); -} - -function isExternalUserText(probe: SessionUpstreamProbe, text: string | undefined): boolean { - const normalized = text === undefined ? "" : normalizeUserText(text); - return !probe.ownRecentUserTexts.includes(normalized); -} - async function checkClaudeSessionUpstreamActivity( probe: SessionUpstreamProbe, ): Promise { diff --git a/extensions/opencode/session-catalog-plugin.ts b/extensions/opencode/session-catalog-plugin.ts index 8382c1a64651..3b2aa235c931 100644 --- a/extensions/opencode/session-catalog-plugin.ts +++ b/extensions/opencode/session-catalog-plugin.ts @@ -46,6 +46,10 @@ import { readLocalOpenCodeTranscriptPage, type OpenCodeSessionPage, } from "./session-catalog.js"; +import { + checkOpenCodeUpstreamActivity, + linkContinuedOpenCodeSession, +} from "./session-upstream-activity.js"; const MAX_HOSTS = 100; const TRANSCRIPT_ITEM_TYPES = new Set([ @@ -60,7 +64,10 @@ const ACPX_BACKEND_ID = "acpx"; const OPENCODE_ACP_AGENT_ID = "opencode"; const OPENCODE_ADOPTED_SESSION_KEY_PREFIX = "plugin:opencode:catalog-adopt:"; -const continueAdoption = createSessionCatalogAdoptionCoordinator(); +const continueAdoption = + createSessionCatalogAdoptionCoordinator< + Awaited> + >(); function isOptionalString(value: unknown): boolean { return value === undefined || typeof value === "string"; @@ -456,7 +463,7 @@ async function continueOpenCodeSession( api: OpenClawPluginApi, hostId: string, threadId: string, -): Promise<{ sessionKey: string }> { +): Promise>> { if (hostId.startsWith("node:")) { throw new OpenCodeCatalogParamsError("paired-node OpenCode session rows are view-only"); } @@ -517,6 +524,8 @@ async function continueOpenCodeSession( }); return { sessionKey: created.key }; }, + complete: async (continued) => + await linkContinuedOpenCodeSession(continued.sessionKey, threadId), }); } @@ -531,6 +540,7 @@ export function registerOpenCodeSessionCatalog(api: OpenClawPluginApi): void { read: async (request) => await readOpenCodeTranscript(api.runtime, request), continueSession: async (request) => await continueOpenCodeSession(api, request.hostId, request.threadId), + checkUpstreamActivity: checkOpenCodeUpstreamActivity, openTerminal: async (request) => await openOpenCodeCatalogTerminal({ runtime: api.runtime, diff --git a/extensions/opencode/session-catalog.test.ts b/extensions/opencode/session-catalog.test.ts index 93340276dc7b..37da0c7539ec 100644 --- a/extensions/opencode/session-catalog.test.ts +++ b/extensions/opencode/session-catalog.test.ts @@ -197,7 +197,16 @@ async function installFakeOpenCode( const args = process.argv.slice(2); if (process.env.CATALOG_UNRELATED_ENV) process.exit(3); if (args[0] === "--pure" && args[1] === "db" && args.includes("--format") && args.includes("json")) { - process.stdout.write(${JSON.stringify(JSON.stringify([session]))}); + process.stdout.write(args[2].includes("event_sequence") + ? ${JSON.stringify( + JSON.stringify([ + { + id: "ses_test", + seq: 4, + }, + ]), + )} + : ${JSON.stringify(JSON.stringify([session]))}); } else if (args[0] === "--pure" && args[1] === "export" && args[2] === "ses_test") { process.stdout.write(${JSON.stringify(JSON.stringify(exported))}); } else { @@ -419,6 +428,14 @@ describe("OpenCode session catalog", () => { expect(first).toEqual(concurrent); expect(second).toEqual(first); + expect(first.upstream).toEqual({ + kind: "opencode-cli", + ref: { threadId: "ses_test" }, + marker: { + seq: 4, + lastHumanMessageId: "msg_user", + }, + }); expect(createSessionEntry).toHaveBeenCalledTimes(1); expect(createSessionEntry).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/extensions/opencode/session-catalog.ts b/extensions/opencode/session-catalog.ts index e2049057cf40..7b3ec1fd30fa 100644 --- a/extensions/opencode/session-catalog.ts +++ b/extensions/opencode/session-catalog.ts @@ -302,6 +302,16 @@ async function runOpenCode(args: string[]): Promise { return Buffer.concat(stdout).toString("utf8"); } +export async function queryOpenCodeDatabase(query: string): Promise { + const output = await runOpenCode(["--pure", "db", query, "--format", "json"]); + return output.trim() ? (JSON.parse(output) as unknown) : []; +} + +export async function exportOpenCodeSession(threadId: string): Promise { + const output = await runOpenCode(["--pure", "export", threadId]); + return JSON.parse(output) as unknown; +} + function parseOpenCodeSession(value: unknown): SessionCatalogSession | undefined { if (!isRecord(value)) { return undefined; @@ -343,8 +353,7 @@ export async function listLocalOpenCodeSessionPage(value?: unknown): Promise MAX_CLI_LIST_SESSIONS) { throw new Error("OpenCode returned an invalid session list"); } @@ -478,8 +487,7 @@ export async function readLocalOpenCodeTranscriptPage( ): Promise { const params = parseReadParams(value); const offset = decodeCursor(params.cursor); - const output = await runOpenCode(["--pure", "export", params.threadId]); - const items = openCodeTranscriptItems(JSON.parse(output) as unknown); + const items = openCodeTranscriptItems(await exportOpenCodeSession(params.threadId)); const page = transcriptPage(items, params.limit, offset); return { hostId: LOCAL_HOST_ID, diff --git a/extensions/opencode/session-upstream-activity.test.ts b/extensions/opencode/session-upstream-activity.test.ts new file mode 100644 index 000000000000..3509765283ea --- /dev/null +++ b/extensions/opencode/session-upstream-activity.test.ts @@ -0,0 +1,653 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + checkOpenCodeUpstreamActivity, + linkContinuedOpenCodeSession, +} from "./session-upstream-activity.js"; + +type StatefulOpenCodeSession = { + id: string; + title: string; + directory: string; + seq?: number; + messages: Array<{ + info: { id: string; role: string; time: { created: number } }; + parts: Array<{ + id: string; + type: string; + text?: string; + synthetic?: boolean; + ignored?: boolean; + metadata?: Record; + mime?: string; + filename?: string; + source?: { text: { value: string; start: number; end: number } }; + }>; + }>; +}; + +type Probe = Parameters[0][number]; + +const temporaryDirectories: string[] = []; +const originalPath = process.env.PATH; + +afterEach(async () => { + process.env.PATH = originalPath; + await Promise.all( + temporaryDirectories.splice(0).map(async (directory) => { + await fs.rm(directory, { recursive: true, force: true }); + }), + ); +}); + +async function installStatefulOpenCode(initialSessions: StatefulOpenCodeSession[]) { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-opencode-activity-")); + temporaryDirectories.push(directory); + const executable = path.join(directory, "opencode"); + const stateFile = path.join(directory, "state.json"); + const logFile = path.join(directory, "calls.jsonl"); + const activeExportsDirectory = path.join(directory, "active-exports"); + const exportConcurrencyLog = path.join(directory, "export-concurrency.log"); + await fs.mkdir(activeExportsDirectory); + const writeState = async (state: { + sessions: StatefulOpenCodeSession[]; + failDb?: boolean; + failExports?: string[]; + exportDelayMs?: number; + }) => await fs.writeFile(stateFile, JSON.stringify(state)); + await writeState({ sessions: initialSessions }); + await fs.writeFile(logFile, ""); + await fs.writeFile( + executable, + "#!/usr/bin/env node\n" + + `const fs = require("node:fs"); +const args = process.argv.slice(2); +fs.appendFileSync(${JSON.stringify(logFile)}, JSON.stringify(args) + "\\n"); +const state = JSON.parse(fs.readFileSync(${JSON.stringify(stateFile)}, "utf8")); +if (args[0] === "--pure" && args[1] === "db") { + if (state.failDb) process.exit(4); + const query = args[2]; + const selected = state.sessions.filter((session) => query.includes("'" + session.id + "'")); + process.stdout.write(JSON.stringify(selected.map((session) => ({ + id: session.id, + seq: session.seq ?? null, + })))); +} else if (args[0] === "--pure" && args[1] === "export") { + const session = state.sessions.find((candidate) => candidate.id === args[2]); + if (!session || state.failExports?.includes(args[2])) process.exit(5); + const activeMarker = ${JSON.stringify(activeExportsDirectory)} + "/" + process.pid; + fs.writeFileSync(activeMarker, ""); + fs.appendFileSync( + ${JSON.stringify(exportConcurrencyLog)}, + String(fs.readdirSync(${JSON.stringify(activeExportsDirectory)}).length) + "\\n", + ); + const finish = () => { + fs.rmSync(activeMarker, { force: true }); + process.stdout.write(JSON.stringify({ info: session, messages: session.messages })); + }; + if (state.exportDelayMs) setTimeout(finish, state.exportDelayMs); + else finish(); +} else { + process.exit(2); +} +`, + ); + await fs.chmod(executable, 0o755); + process.env.PATH = `${directory}${path.delimiter}${originalPath ?? ""}`; + return { + writeState, + clearLog: async () => await fs.writeFile(logFile, ""), + readCalls: async () => + (await fs.readFile(logFile, "utf8")) + .trim() + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line) as string[]), + readMaxExportConcurrency: async () => + Math.max( + 0, + ...(await fs.readFile(exportConcurrencyLog, "utf8").catch(() => "")) + .trim() + .split("\n") + .filter(Boolean) + .map(Number), + ), + }; +} + +function openCodeMessage(id: string, role: string, text: string, created: number) { + return { + info: { id, role, time: { created } }, + parts: [{ id: `part-${id}`, type: "text", text }], + }; +} + +function probe(marker: Probe["marker"], ownRecentUserTexts: string[] = []): Probe { + return { + sessionKey: "agent:main:ses-a", + agentId: "main", + threadId: "ses_a", + hostId: "gateway", + upstreamKind: "opencode-cli", + upstreamRef: { threadId: "ses_a" }, + marker, + ownRecentUserTexts, + }; +} + +function markerFrom( + outcome: Awaited>[number] | undefined, +) { + if (outcome?.kind !== "activity") { + throw new Error("expected activity marker"); + } + return outcome.nextMarker; +} + +describe("OpenCode session upstream activity", () => { + it.runIf(process.platform !== "win32")( + "uses event_sequence and exports only after the cursor advances", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 1, + messages: [openCodeMessage("msg_001", "assistant", "ready", 1_700_000_000_000)], + }; + const fixture = await installStatefulOpenCode([session]); + const continued = await linkContinuedOpenCodeSession("agent:main:ses-a", "ses_a"); + expect(continued.upstream?.marker).toEqual({ seq: 1, lastHumanMessageId: null }); + + await fixture.clearLog(); + await expect( + checkOpenCodeUpstreamActivity([probe(continued.upstream!.marker)]), + ).resolves.toEqual([]); + let calls = await fixture.readCalls(); + expect(calls).toHaveLength(1); + expect(calls[0]?.[2]).toBe( + "SELECT s.id AS id, es.seq AS seq FROM session AS s LEFT JOIN event_sequence AS es ON es.aggregate_id = s.id WHERE s.id IN ('ses_a')", + ); + + session.messages.push( + openCodeMessage("msg_002", "user", "external OpenCode turn", 1_700_000_001_000), + ); + session.seq = 3; + await fixture.writeState({ sessions: [session] }); + await fixture.clearLog(); + await expect( + checkOpenCodeUpstreamActivity([probe(continued.upstream!.marker)]), + ).resolves.toEqual([ + expect.objectContaining({ + kind: "activity", + humanTurns: 1, + dedupeId: "msg_002", + nextMarker: { seq: 3, lastHumanMessageId: "msg_002" }, + }), + ]); + calls = await fixture.readCalls(); + expect(calls.filter((args) => args[1] === "db")).toHaveLength(1); + expect(calls.filter((args) => args[1] === "export")).toHaveLength(1); + }, + ); + + it.runIf(process.platform !== "win32")( + "re-baselines a regressed cursor and keeps monitoring the next advance", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 0, + messages: [openCodeMessage("msg_001", "user", "old turn", 1_700_000_000_000)], + }; + const fixture = await installStatefulOpenCode([session]); + const reset = await checkOpenCodeUpstreamActivity([ + probe({ seq: 12, lastHumanMessageId: "msg_001" }), + ]); + expect(reset).toEqual([ + { + kind: "activity", + sessionKey: "agent:main:ses-a", + humanTurns: 0, + nextMarker: { seq: 0, lastHumanMessageId: "msg_001" }, + }, + ]); + expect((await fixture.readCalls()).filter((args) => args[1] === "export")).toHaveLength(0); + + session.messages.push( + openCodeMessage("msg_002", "user", "new after migration", 1_700_000_001_000), + ); + session.seq = 2; + await fixture.writeState({ sessions: [session] }); + await expect(checkOpenCodeUpstreamActivity([probe(markerFrom(reset[0]))])).resolves.toEqual([ + expect.objectContaining({ + humanTurns: 1, + nextMarker: { seq: 2, lastHumanMessageId: "msg_002" }, + }), + ]); + }, + ); + + it.runIf(process.platform !== "win32")( + "waits across staged part projections and reports the human turn exactly once", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 1, + messages: [openCodeMessage("msg_001", "assistant", "ready", 1_700_000_000_000)], + }; + const fixture = await installStatefulOpenCode([session]); + let currentMarker: Probe["marker"] = { seq: 1, lastHumanMessageId: null }; + const user = { + info: { id: "msg_002", role: "user", time: { created: 1_700_000_001_000 } }, + parts: [] as StatefulOpenCodeSession["messages"][number]["parts"], + }; + session.messages.push(user); + session.seq = 2; + await fixture.writeState({ sessions: [session] }); + let outcomes = await checkOpenCodeUpstreamActivity([probe(currentMarker)]); + expect(outcomes).toEqual([expect.objectContaining({ humanTurns: 0 })]); + currentMarker = markerFrom(outcomes[0]); + + user.parts.push({ id: "part-file", type: "file" }); + session.seq = 3; + await fixture.writeState({ sessions: [session] }); + outcomes = await checkOpenCodeUpstreamActivity([probe(currentMarker)]); + expect(outcomes).toEqual([expect.objectContaining({ humanTurns: 0 })]); + currentMarker = markerFrom(outcomes[0]); + + user.parts.push({ id: "part-text", type: "text", text: "arrived in stages" }); + session.seq = 4; + await fixture.writeState({ sessions: [session] }); + outcomes = await checkOpenCodeUpstreamActivity([probe(currentMarker)]); + expect(outcomes).toEqual([ + expect.objectContaining({ + humanTurns: 1, + nextMarker: { seq: 4, lastHumanMessageId: "msg_002" }, + }), + ]); + currentMarker = markerFrom(outcomes[0]); + + user.parts.push({ id: "part-late", type: "text", text: "late detail" }); + session.seq = 5; + await fixture.writeState({ sessions: [session] }); + await expect(checkOpenCodeUpstreamActivity([probe(currentMarker)])).resolves.toEqual([ + expect.objectContaining({ + humanTurns: 0, + nextMarker: { seq: 5, lastHumanMessageId: "msg_002" }, + }), + ]); + }, + ); + + it.runIf(process.platform !== "win32")( + "suppresses compaction replay text duplicated from an earlier user turn", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 9, + messages: [ + { + info: { id: "msg_001", role: "user", time: { created: 1_700_000_000_000 } }, + parts: [ + { id: "part-hidden", type: "text", text: "hidden", ignored: true }, + { id: "part-visible", type: "text", text: "please keep going" }, + ], + }, + openCodeMessage("msg_002", "assistant", "working", 1_700_000_001_000), + { + info: { id: "msg_003", role: "user", time: { created: 1_700_000_002_000 } }, + parts: [{ id: "part-compact", type: "compaction" }], + }, + openCodeMessage("msg_004", "assistant", "summary", 1_700_000_003_000), + openCodeMessage("msg_005", "user", " please keep going ", 1_700_000_004_000), + ], + }; + await installStatefulOpenCode([session]); + await expect( + checkOpenCodeUpstreamActivity([probe({ seq: 4, lastHumanMessageId: "msg_001" })]), + ).resolves.toEqual([ + { + kind: "activity", + sessionKey: "agent:main:ses-a", + humanTurns: 0, + nextMarker: { seq: 9, lastHumanMessageId: "msg_005" }, + }, + ]); + }, + ); + + it.runIf(process.platform !== "win32")( + "normalizes media placeholders when suppressing compaction replay", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 7, + messages: [ + { + info: { id: "msg_001", role: "user", time: { created: 1_700_000_000_000 } }, + parts: [ + { + id: "part-image", + type: "file", + mime: "image/png", + filename: "screen.png", + }, + ], + }, + openCodeMessage("msg_002", "assistant", "working", 1_700_000_001_000), + { + info: { id: "msg_003", role: "user", time: { created: 1_700_000_002_000 } }, + parts: [{ id: "part-compact", type: "compaction" }], + }, + openCodeMessage("msg_004", "assistant", "summary", 1_700_000_003_000), + openCodeMessage("msg_005", "user", "[Attached image/png: screen.png]", 1_700_000_004_000), + ], + }; + await installStatefulOpenCode([session]); + await expect( + checkOpenCodeUpstreamActivity([probe({ seq: 2, lastHumanMessageId: null })]), + ).resolves.toEqual([ + { + kind: "activity", + sessionKey: "agent:main:ses-a", + humanTurns: 0, + nextMarker: { seq: 7, lastHumanMessageId: "msg_005" }, + }, + ]); + }, + ); + + it.runIf(process.platform !== "win32")( + "bounds replay suppression to the preceding 50 user messages", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 80, + messages: [ + openCodeMessage("msg_001", "user", "repeat me", 1_700_000_000_000), + ...Array.from({ length: 50 }, (_, index) => ({ + info: { + id: `msg_${String(index + 2).padStart(3, "0")}`, + role: "user", + time: { created: 1_700_000_001_000 + index }, + }, + parts: [ + { + id: `part-${String(index)}`, + type: "text", + text: "internal", + synthetic: true, + }, + ], + })), + openCodeMessage("msg_052", "user", "repeat me", 1_700_000_052_000), + ], + }; + await installStatefulOpenCode([session]); + await expect( + checkOpenCodeUpstreamActivity([probe({ seq: 1, lastHumanMessageId: "msg_001" })]), + ).resolves.toEqual([ + expect.objectContaining({ + humanTurns: 1, + nextMarker: { seq: 80, lastHumanMessageId: "msg_052" }, + }), + ]); + }, + ); + + it.runIf(process.platform !== "win32")( + "dedupes SessionSummary re-publication of an existing human message id", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 7, + messages: [openCodeMessage("msg_001", "user", "already reported", 1_700_000_000_000)], + }; + const fixture = await installStatefulOpenCode([session]); + const outcomes = await checkOpenCodeUpstreamActivity([ + probe({ seq: 6, lastHumanMessageId: "msg_001" }), + ]); + expect(outcomes).toEqual([ + { + kind: "activity", + sessionKey: "agent:main:ses-a", + humanTurns: 0, + nextMarker: { seq: 7, lastHumanMessageId: "msg_001" }, + }, + ]); + await fixture.clearLog(); + await expect( + checkOpenCodeUpstreamActivity([probe(markerFrom(outcomes[0]))]), + ).resolves.toEqual([]); + expect((await fixture.readCalls()).filter((args) => args[1] === "export")).toHaveLength(0); + }, + ); + + it.runIf(process.platform !== "win32")( + "suppresses ignored, synthetic, shell, compaction, and continuation user rows", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 12, + messages: [ + openCodeMessage("msg_001", "assistant", "ready", 1_700_000_000_000), + { + info: { id: "msg_002", role: "user", time: { created: 1_700_000_001_000 } }, + parts: [{ id: "part-ignored", type: "text", text: "hidden", ignored: true }], + }, + { + info: { id: "msg_003", role: "user", time: { created: 1_700_000_002_000 } }, + parts: [{ id: "part-synthetic", type: "text", text: "hidden", synthetic: true }], + }, + openCodeMessage( + "msg_004", + "user", + "The following tool was executed by the user", + 1_700_000_003_000, + ), + { + info: { id: "msg_005", role: "user", time: { created: 1_700_000_004_000 } }, + parts: [ + { id: "part-text", type: "text", text: "looks human" }, + { id: "part-compaction", type: "compaction" }, + ], + }, + { + info: { id: "msg_006", role: "user", time: { created: 1_700_000_005_000 } }, + parts: [ + { + id: "part-continue", + type: "text", + text: "continue", + metadata: { compaction_continue: true }, + }, + ], + }, + ], + }; + await installStatefulOpenCode([session]); + await expect( + checkOpenCodeUpstreamActivity([probe({ seq: 1, lastHumanMessageId: null })]), + ).resolves.toEqual([ + { + kind: "activity", + sessionKey: "agent:main:ses-a", + humanTurns: 0, + nextMarker: { seq: 12, lastHumanMessageId: null }, + }, + ]); + }, + ); + + it.runIf(process.platform !== "win32")("does not report a file-mention-only turn", async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 3, + messages: [ + openCodeMessage("msg_001", "assistant", "ready", 1_700_000_000_000), + { + info: { id: "msg_002", role: "user", time: { created: 1_700_000_001_000 } }, + parts: [ + { id: "part-text", type: "text", text: "@notes.ts" }, + { + id: "part-file", + type: "file", + mime: "text/plain", + filename: "notes.ts", + source: { text: { value: "@notes.ts", start: 0, end: 9 } }, + }, + ], + }, + ], + }; + await installStatefulOpenCode([session]); + await expect( + checkOpenCodeUpstreamActivity([probe({ seq: 1, lastHumanMessageId: null })]), + ).resolves.toEqual([ + { + kind: "activity", + sessionKey: "agent:main:ses-a", + humanTurns: 0, + nextMarker: { seq: 3, lastHumanMessageId: null }, + }, + ]); + }); + + it.runIf(process.platform !== "win32")( + "keeps real text mixed with ignored text and suppresses OpenClaw self-echo", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 4, + messages: [ + { + info: { id: "msg_001", role: "user", time: { created: 1_700_000_001_000 } }, + parts: [ + { id: "part-hidden", type: "text", text: "hidden", ignored: true }, + { id: "part-real", type: "text", text: "real external turn" }, + ], + }, + ], + }; + const fixture = await installStatefulOpenCode([session]); + await expect( + checkOpenCodeUpstreamActivity([probe({ seq: 0, lastHumanMessageId: null })]), + ).resolves.toEqual([expect.objectContaining({ humanTurns: 1 })]); + await expect( + checkOpenCodeUpstreamActivity([ + probe({ seq: 0, lastHumanMessageId: null }, ["real external turn"]), + ]), + ).resolves.toEqual([ + expect.objectContaining({ + humanTurns: 0, + nextMarker: { seq: 4, lastHumanMessageId: "msg_001" }, + }), + ]); + + session.messages.push(openCodeMessage("msg_002", "assistant", "summary", 1_700_000_002_000)); + session.seq = 5; + await fixture.writeState({ sessions: [session] }); + const suppressed = await checkOpenCodeUpstreamActivity([ + probe({ seq: 4, lastHumanMessageId: "msg_001" }), + ]); + expect(suppressed).toEqual([expect.objectContaining({ humanTurns: 0 })]); + }, + ); + + it.runIf(process.platform !== "win32")( + "reports confirmed absence but not query or export failures", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + seq: 2, + messages: [openCodeMessage("msg_001", "user", "external", 1_700_000_001_000)], + }; + const fixture = await installStatefulOpenCode([session]); + const currentProbe = probe({ seq: 1, lastHumanMessageId: null }); + + await fixture.writeState({ sessions: [], failDb: true }); + await expect(checkOpenCodeUpstreamActivity([currentProbe])).resolves.toEqual([]); + + await fixture.writeState({ sessions: [session], failExports: ["ses_a"] }); + await expect(checkOpenCodeUpstreamActivity([currentProbe])).resolves.toEqual([]); + + await fixture.writeState({ sessions: [] }); + await expect(checkOpenCodeUpstreamActivity([currentProbe])).resolves.toEqual([ + { kind: "missing", sessionKey: "agent:main:ses-a" }, + ]); + }, + ); + + it.runIf(process.platform !== "win32")( + "treats a missing event_sequence row as sequence zero", + async () => { + const session: StatefulOpenCodeSession = { + id: "ses_a", + title: "Session A", + directory: "/workspace/a", + messages: [], + }; + await installStatefulOpenCode([session]); + await expect(linkContinuedOpenCodeSession("agent:main:ses-a", "ses_a")).resolves.toEqual({ + sessionKey: "agent:main:ses-a", + upstream: { + kind: "opencode-cli", + ref: { threadId: "ses_a" }, + marker: { seq: 0, lastHumanMessageId: null }, + }, + }); + }, + ); + + it.runIf(process.platform !== "win32")( + "bounds concurrent exports when many cursors advance", + async () => { + const sessions: StatefulOpenCodeSession[] = Array.from({ length: 9 }, (_, index) => ({ + id: `ses_${String(index)}`, + title: `Session ${String(index)}`, + directory: `/workspace/${String(index)}`, + seq: 2, + messages: [ + openCodeMessage(`msg_${String(index)}`, "assistant", "ready", 1_700_000_001_000 + index), + ], + })); + const fixture = await installStatefulOpenCode(sessions); + await fixture.writeState({ sessions, exportDelayMs: 250 }); + + await expect( + checkOpenCodeUpstreamActivity( + sessions.map((session) => ({ + ...probe({ seq: 1, lastHumanMessageId: null }), + sessionKey: `agent:main:${session.id}`, + threadId: session.id, + upstreamRef: { threadId: session.id }, + })), + ), + ).resolves.toHaveLength(sessions.length); + expect(await fixture.readMaxExportConcurrency()).toBeGreaterThan(0); + expect(await fixture.readMaxExportConcurrency()).toBeLessThanOrEqual(4); + }, + ); +}); diff --git a/extensions/opencode/session-upstream-activity.ts b/extensions/opencode/session-upstream-activity.ts new file mode 100644 index 000000000000..14cad0857c08 --- /dev/null +++ b/extensions/opencode/session-upstream-activity.ts @@ -0,0 +1,429 @@ +import { + isExternalUserText, + normalizeUserText, + type SessionCatalogContinueProviderResult, + type SessionUpstreamActivity, + type SessionUpstreamProbe, +} from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { OPENCODE_SESSION_ID_PATTERN } from "./session-catalog-shared.js"; +import { exportOpenCodeSession, queryOpenCodeDatabase } from "./session-catalog.js"; + +type OpenCodeIndicator = { + threadId: string; + seq: number; +}; +type OpenCodeExportPart = { + type: string; + text?: string; + synthetic?: boolean; + ignored?: boolean; + metadata?: Record; + mime?: string; + filename?: string; + sourceText?: { + value: string; + start: number; + end: number; + }; +}; +type OpenCodeExportMessage = { + id: string; + role: string; + parts: OpenCodeExportPart[]; + createdAt?: number; +}; +type OpenCodeMarker = { + seq: number; + lastHumanMessageId: string | null; +}; + +const OPENCODE_EXPORT_CONCURRENCY = 4; +const OPENCODE_REPLAY_LOOKBACK_USER_MESSAGES = 50; +const OPENCODE_SHELL_SENTINEL = "The following tool was executed by the user"; + +async function mapConcurrent( + values: T[], + limit: number, + mapper: (value: T) => Promise, +): Promise { + const results: R[] = []; + results.length = values.length; + let nextIndex = 0; + const workers = Array.from({ length: Math.min(limit, values.length) }, async () => { + while (nextIndex < values.length) { + const index = nextIndex++; + results[index] = await mapper(values[index]!); + } + }); + await Promise.all(workers); + return results; +} + +function sqlString(value: string): string { + return `'${value.replaceAll("'", "''")}'`; +} + +function readProbeThreadId(probe: SessionUpstreamProbe): string | undefined { + if ( + probe.hostId !== "gateway" || + probe.upstreamKind !== "opencode-cli" || + !isRecord(probe.upstreamRef) || + probe.upstreamRef.threadId !== probe.threadId || + !OPENCODE_SESSION_ID_PATTERN.test(probe.threadId) + ) { + return undefined; + } + return probe.threadId; +} + +function readMarker(probe: SessionUpstreamProbe): OpenCodeMarker | undefined { + if (!isRecord(probe.marker)) { + return undefined; + } + return Number.isSafeInteger(probe.marker.seq) && + Number(probe.marker.seq) >= 0 && + (probe.marker.lastHumanMessageId === null || + typeof probe.marker.lastHumanMessageId === "string") + ? { + seq: Number(probe.marker.seq), + lastHumanMessageId: probe.marker.lastHumanMessageId, + } + : undefined; +} + +async function readIndicators(threadIds: string[]): Promise> { + if (threadIds.length === 0) { + return new Map(); + } + const query = [ + "SELECT s.id AS id, es.seq AS seq", + "FROM session AS s", + "LEFT JOIN event_sequence AS es ON es.aggregate_id = s.id", + `WHERE s.id IN (${threadIds.map(sqlString).join(", ")})`, + ].join(" "); + const value = await queryOpenCodeDatabase(query); + if (!Array.isArray(value)) { + throw new Error("OpenCode returned invalid upstream indicators"); + } + const indicators = new Map(); + for (const row of value) { + const seq = isRecord(row) && row.seq === null ? 0 : isRecord(row) ? row.seq : undefined; + if ( + !isRecord(row) || + typeof row.id !== "string" || + !OPENCODE_SESSION_ID_PATTERN.test(row.id) || + !Number.isSafeInteger(seq) || + Number(seq) < 0 + ) { + throw new Error("OpenCode returned invalid upstream indicators"); + } + indicators.set(row.id, { threadId: row.id, seq: Number(seq) }); + } + return indicators; +} + +function readExportPart(value: unknown): OpenCodeExportPart | undefined { + if (!isRecord(value) || typeof value.type !== "string") { + return undefined; + } + const sourceText = + isRecord(value.source) && + isRecord(value.source.text) && + typeof value.source.text.value === "string" && + typeof value.source.text.start === "number" && + Number.isFinite(value.source.text.start) && + typeof value.source.text.end === "number" && + Number.isFinite(value.source.text.end) + ? { + value: value.source.text.value, + start: value.source.text.start, + end: value.source.text.end, + } + : undefined; + return { + type: value.type, + ...(typeof value.text === "string" ? { text: value.text } : {}), + ...(typeof value.synthetic === "boolean" ? { synthetic: value.synthetic } : {}), + ...(typeof value.ignored === "boolean" ? { ignored: value.ignored } : {}), + ...(isRecord(value.metadata) ? { metadata: value.metadata } : {}), + ...(typeof value.mime === "string" ? { mime: value.mime } : {}), + ...(typeof value.filename === "string" ? { filename: value.filename } : {}), + ...(sourceText ? { sourceText } : {}), + }; +} + +function visibleTextPart( + part: OpenCodeExportPart, + sourceRanges: NonNullable[], +): string | undefined { + if ( + part.type !== "text" || + part.text === undefined || + part.synthetic === true || + part.ignored === true || + part.metadata?.compaction_continue === true + ) { + return undefined; + } + let text = part.text; + for (const source of sourceRanges) { + if ( + Number.isInteger(source.start) && + Number.isInteger(source.end) && + source.start >= 0 && + source.end >= source.start && + source.end <= text.length && + text.slice(source.start, source.end) === source.value + ) { + text = text.slice(0, source.start) + text.slice(source.end); + } + } + return text; +} + +function messageSourceRanges( + message: OpenCodeExportMessage, +): NonNullable[] { + return message.parts + .flatMap((part) => (part.sourceText ? [part.sourceText] : [])) + .toSorted((left, right) => right.start - left.start); +} + +function visibleTextParts(message: OpenCodeExportMessage): string[] { + const sourceRanges = messageSourceRanges(message); + return message.parts.flatMap((part) => { + const text = visibleTextPart(part, sourceRanges); + return text === undefined ? [] : [text]; + }); +} + +function readExportMessages(value: unknown): OpenCodeExportMessage[] { + if (!isRecord(value) || !Array.isArray(value.messages)) { + throw new Error("OpenCode returned an invalid session export"); + } + return value.messages + .flatMap((message): OpenCodeExportMessage[] => { + if (!isRecord(message) || !isRecord(message.info) || !Array.isArray(message.parts)) { + return []; + } + const id = message.info.id; + const role = message.info.role; + if (typeof id !== "string" || typeof role !== "string") { + return []; + } + const createdAt = + isRecord(message.info.time) && + typeof message.info.time.created === "number" && + Number.isFinite(message.info.time.created) + ? message.info.time.created + : undefined; + return [ + { + id, + role, + parts: message.parts.flatMap((part) => { + const parsed = readExportPart(part); + return parsed ? [parsed] : []; + }), + ...(createdAt === undefined ? {} : { createdAt }), + }, + ]; + }) + .toSorted( + (left, right) => + (left.createdAt ?? Number.NEGATIVE_INFINITY) - + (right.createdAt ?? Number.NEGATIVE_INFINITY) || left.id.localeCompare(right.id), + ); +} + +function normalizedMessageText(message: OpenCodeExportMessage): string | undefined { + if (message.role !== "user") { + return undefined; + } + const sourceRanges = messageSourceRanges(message); + const texts = message.parts.flatMap((part) => { + const visibleText = visibleTextPart(part, sourceRanges); + if (visibleText !== undefined) { + return [visibleText]; + } + if ( + part.type === "file" && + part.mime !== undefined && + (part.mime.startsWith("image/") || part.mime === "application/pdf") + ) { + return [`[Attached ${part.mime}: ${part.filename ?? "file"}]`]; + } + return []; + }); + return texts.length > 0 ? normalizeUserText(texts.join("\n")) : undefined; +} + +function directHumanText(message: OpenCodeExportMessage): string | undefined { + if ( + message.role !== "user" || + message.parts.some((part) => part.type === "compaction") || + message.parts.some((part) => part.metadata?.compaction_continue === true) + ) { + return undefined; + } + const texts = visibleTextParts(message); + if (texts.length === 0) { + return undefined; + } + const text = normalizeUserText(texts.join("\n")); + return !text || text === OPENCODE_SHELL_SENTINEL ? undefined : text; +} + +function latestMessageId(current: string | null, candidate: string): string { + return current === null || candidate > current ? candidate : current; +} + +function latestBaselineHumanMessageId(messages: OpenCodeExportMessage[]): string | null { + let latest: string | null = null; + for (const message of messages) { + const text = directHumanText(message); + if (text !== undefined) { + latest = latestMessageId(latest, message.id); + } + } + return latest; +} + +function classifyExport(params: { + probe: SessionUpstreamProbe; + marker: OpenCodeMarker; + seq: number; + messages: OpenCodeExportMessage[]; +}): SessionUpstreamActivity { + let humanTurns = 0; + let occurredAt: number | undefined; + let lastHumanMessageId = params.marker.lastHumanMessageId; + let latestExternalMessageId: string | undefined; + const earlierUserTexts: Array = []; + for (const message of params.messages) { + const text = directHumanText(message); + const replay = + text !== undefined && + earlierUserTexts.slice(-OPENCODE_REPLAY_LOOKBACK_USER_MESSAGES).includes(text); + const newerThanMarker = + params.marker.lastHumanMessageId === null || message.id > params.marker.lastHumanMessageId; + if (text !== undefined && message.createdAt !== undefined && newerThanMarker) { + // Consume complete user-shaped rows even when replay/self-echo filtering suppresses + // them, so a later own-text window cannot turn an old row into new activity. + lastHumanMessageId = latestMessageId(lastHumanMessageId, message.id); + } + if ( + text !== undefined && + !replay && + message.createdAt !== undefined && + newerThanMarker && + isExternalUserText(params.probe, text) + ) { + humanTurns += 1; + occurredAt = Math.max(occurredAt ?? 0, message.createdAt); + latestExternalMessageId = message.id; + } + if (message.role === "user") { + earlierUserTexts.push(normalizedMessageText(message)); + if (earlierUserTexts.length > OPENCODE_REPLAY_LOOKBACK_USER_MESSAGES) { + earlierUserTexts.shift(); + } + } + } + const nextMarker = { seq: params.seq, lastHumanMessageId }; + return { + kind: "activity", + sessionKey: params.probe.sessionKey, + humanTurns, + nextMarker, + ...(humanTurns > 0 + ? { + occurredAt: occurredAt ?? Date.now(), + dedupeId: latestExternalMessageId ?? String(params.seq), + } + : {}), + }; +} + +export async function linkContinuedOpenCodeSession( + sessionKey: string, + threadId: string, +): Promise { + try { + const indicator = (await readIndicators([threadId])).get(threadId); + if (!indicator) { + return { sessionKey }; + } + const messages = readExportMessages(await exportOpenCodeSession(threadId)); + return { + sessionKey, + upstream: { + kind: "opencode-cli", + ref: { threadId }, + marker: { + seq: indicator.seq, + lastHumanMessageId: latestBaselineHumanMessageId(messages), + }, + }, + }; + } catch { + // Liveness metadata is optional; continuation success must survive baseline failure. + return { sessionKey }; + } +} + +async function classifyChangedProbe( + probe: SessionUpstreamProbe, + indicator: OpenCodeIndicator, +): Promise { + const marker = readMarker(probe); + if (!marker || indicator.seq === marker.seq) { + return undefined; + } + if (indicator.seq < marker.seq) { + // OpenCode migrations may reset event_sequence while preserving the session. + return { + kind: "activity", + sessionKey: probe.sessionKey, + humanTurns: 0, + nextMarker: { seq: indicator.seq, lastHumanMessageId: marker.lastHumanMessageId }, + }; + } + return classifyExport({ + probe, + marker, + seq: indicator.seq, + messages: readExportMessages(await exportOpenCodeSession(probe.threadId)), + }); +} + +export async function checkOpenCodeUpstreamActivity( + probes: SessionUpstreamProbe[], +): Promise { + const eligible = probes.flatMap((probe) => (readProbeThreadId(probe) ? [probe] : [])); + let indicators: Map; + try { + indicators = await readIndicators([...new Set(eligible.map((probe) => probe.threadId))]); + } catch { + // A failed batch read confirms nothing about whether any thread still exists. + return []; + } + const outcomes = await mapConcurrent( + eligible, + OPENCODE_EXPORT_CONCURRENCY, + async (probe): Promise => { + const indicator = indicators.get(probe.threadId); + if (!indicator) { + return { kind: "missing", sessionKey: probe.sessionKey }; + } + try { + return await classifyChangedProbe(probe, indicator); + } catch { + // Export failures are transient reads, not evidence that a thread was deleted. + return undefined; + } + }, + ); + return outcomes.filter((outcome): outcome is SessionUpstreamActivity => outcome !== undefined); +} diff --git a/src/plugin-sdk/session-catalog.ts b/src/plugin-sdk/session-catalog.ts index 8f440f686e1d..954c1c8c1b2e 100644 --- a/src/plugin-sdk/session-catalog.ts +++ b/src/plugin-sdk/session-catalog.ts @@ -13,7 +13,9 @@ export type { } from "../plugins/session-catalog.js"; export { createSessionCatalogAdoptionCoordinator, + isExternalUserText, listAdoptedSessionCatalogSessions, + normalizeUserText, sessionCatalogAdoptedSessionKey, sessionCatalogAdoptedSourceKey, } from "../plugins/session-catalog.js"; diff --git a/src/plugins/session-catalog.ts b/src/plugins/session-catalog.ts index 11f92bc01448..e7dee9e278e7 100644 --- a/src/plugins/session-catalog.ts +++ b/src/plugins/session-catalog.ts @@ -61,7 +61,7 @@ export type SessionUpstreamJsonValue = | SessionUpstreamJsonValue[] | { [key: string]: SessionUpstreamJsonValue }; -export type SessionUpstreamKind = "claude-cli" | "codex-app-server"; +export type SessionUpstreamKind = "claude-cli" | "codex-app-server" | "opencode-cli" | "pi-cli"; export type SessionUpstreamProbe = { sessionKey: string; @@ -74,6 +74,15 @@ export type SessionUpstreamProbe = { ownRecentUserTexts: string[]; }; +export function normalizeUserText(text: string): string { + return text.trim().replace(/\s+/g, " "); +} + +export function isExternalUserText(probe: SessionUpstreamProbe, text: string | undefined): boolean { + const normalized = text === undefined ? "" : normalizeUserText(text); + return !probe.ownRecentUserTexts.includes(normalized); +} + export type SessionUpstreamActivity = | { kind: "activity"; @@ -164,28 +173,38 @@ export function listAdoptedSessionCatalogSessions(params: { return adopted; } -export function createSessionCatalogAdoptionCoordinator() { - const operations = new Map>(); +// `complete` is intentionally required, not optional-with-fallback: adoption and its +// upstream baseline must share one single-flight operation, or concurrent continues +// race to baseline the same thread. This helper shipped in no release tag yet +// (added #113718), so no external plugin can depend on the older 3-field shape. +export function createSessionCatalogAdoptionCoordinator() { + const operations = new Map>(); return async (params: { sourceKey: string; findExisting: () => string | undefined; create: () => Promise<{ sessionKey: string }>; - }): Promise<{ sessionKey: string }> => { - const existing = params.findExisting(); - if (existing) { - return { sessionKey: existing }; - } + complete: (continued: { sessionKey: string }) => Promise; + }): Promise => { const pending = operations.get(params.sourceKey); if (pending) { return await pending; } - const operation = params.create().catch((error: unknown) => { - const raced = params.findExisting(); - if (raced) { - return { sessionKey: raced }; + const operation = (async () => { + const existing = params.findExisting(); + if (existing) { + // The gateway's same-source link upsert preserves its active marker. Re-running + // completion only supplies a new baseline after that link was removed. + return await params.complete({ sessionKey: existing }); } - throw error; - }); + const continued = await params.create().catch((error: unknown) => { + const raced = params.findExisting(); + if (raced) { + return { sessionKey: raced }; + } + throw error; + }); + return await params.complete(continued); + })(); operations.set(params.sourceKey, operation); try { return await operation;