mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-28 05:16:23 -06:00
feat(cloud-workers): durable transcript commit protocol for worker sessions (#104809)
* feat(cloud-workers): add transcript commit protocol * feat(cloud-workers): persist transcript commit replay state * feat(cloud-workers): apply transcript commits to sessions * fix(cloud-workers): harden transcript commit admission * fix(cloud-workers): recover branched transcript commits * fix(cloud-workers): preserve diagnostic semantics
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
2e6c1090c1
commit
04f945fabe
@@ -176,8 +176,12 @@ verifies a hash-at-rest, short-lived credential bound to the environment, bundle
|
||||
hash, owner epoch, RPC-set version, expiry, and one nullable session; it
|
||||
separately checks the current version and feature set. Success returns minimal
|
||||
`worker-hello-ok`; feature negotiation is independent of the general protocol
|
||||
version. Frames stay under 64 KiB. Initially only `worker.heartbeat` is allowed,
|
||||
with ownership and expiry rechecked on each RPC.
|
||||
version. Frames stay under 64 KiB. The closed allowlist contains
|
||||
`worker.heartbeat` and `worker.transcript.commit` for ordered semantic message batches. Transcript
|
||||
commits use owner-epoch fencing, a gateway-owned session binding, base-leaf
|
||||
compare-and-swap, and durable sequence replay; the gateway generates transcript
|
||||
entry and parent IDs through the normal session writer. Ownership and expiry are
|
||||
rechecked on each RPC.
|
||||
|
||||
### Client capabilities
|
||||
|
||||
|
||||
@@ -421,6 +421,20 @@ import {
|
||||
type WorkerHelloOk,
|
||||
type WorkerProtocolCloseReason,
|
||||
WorkerProtocolCloseReasonSchema,
|
||||
type WorkerTranscriptCommitErrorReason,
|
||||
WorkerTranscriptCommitErrorReasonSchema,
|
||||
type WorkerTranscriptCommitErrorShape,
|
||||
WorkerTranscriptCommitErrorShapeSchema,
|
||||
type WorkerTranscriptCommitParams,
|
||||
WorkerTranscriptCommitParamsSchema,
|
||||
type WorkerTranscriptCommitRequestFrame,
|
||||
WorkerTranscriptCommitRequestFrameSchema,
|
||||
type WorkerTranscriptCommitResponseFrame,
|
||||
WorkerTranscriptCommitResponseFrameSchema,
|
||||
type WorkerTranscriptCommitResult,
|
||||
WorkerTranscriptCommitResultSchema,
|
||||
type WorkerTranscriptMessage,
|
||||
WorkerTranscriptMessageSchema,
|
||||
WORKER_HEARTBEAT_INTERVAL_MS,
|
||||
WORKER_PROTOCOL_FEATURES,
|
||||
WORKER_PROTOCOL_MAX_FEATURE_LENGTH,
|
||||
@@ -431,6 +445,10 @@ import {
|
||||
WORKER_PROTOCOL_MAX_PAYLOAD_BYTES,
|
||||
WORKER_PROTOCOL_METHODS,
|
||||
WORKER_RPC_SET_VERSION,
|
||||
WORKER_TRANSCRIPT_MAX_BATCH_MESSAGES,
|
||||
WORKER_TRANSCRIPT_MAX_CONTENT_PARTS,
|
||||
WORKER_TRANSCRIPT_MAX_JSON_DEPTH,
|
||||
WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE,
|
||||
type SystemInfoParams,
|
||||
SystemInfoParamsSchema,
|
||||
type SystemInfoResult,
|
||||
@@ -861,7 +879,10 @@ export type ProtocolValidator<T = unknown> = ((data: unknown) => data is T) & {
|
||||
|
||||
// Defer TypeBox compilation until the first validation call. Importing this
|
||||
// module is common in CLIs/tests, so eager compilation would add startup cost.
|
||||
function lazyCompile<T = unknown>(schema: unknown): ProtocolValidator<T> {
|
||||
function lazyCompile<T = unknown>(
|
||||
schema: unknown,
|
||||
precheck?: (data: unknown) => ValidationError | undefined,
|
||||
): ProtocolValidator<T> {
|
||||
let compiled: TypeBoxValidator | undefined;
|
||||
let errors: ValidationError[] | null = null;
|
||||
|
||||
@@ -871,6 +892,11 @@ function lazyCompile<T = unknown>(schema: unknown): ProtocolValidator<T> {
|
||||
};
|
||||
|
||||
const validate = ((data: unknown): data is T => {
|
||||
const precheckError = precheck?.(data);
|
||||
if (precheckError) {
|
||||
errors = [precheckError];
|
||||
return false;
|
||||
}
|
||||
const current = getCompiled();
|
||||
const valid = current.Check(data);
|
||||
errors = valid ? null : ([...current.Errors(data)] as ValidationError[]);
|
||||
@@ -910,6 +936,56 @@ export const validateWorkerConnectRequestFrame = lazyCompile<WorkerConnectReques
|
||||
export const validateWorkerHeartbeatParams = lazyCompile<WorkerHeartbeatParams>(
|
||||
WorkerHeartbeatParamsSchema,
|
||||
);
|
||||
|
||||
function checkWorkerTranscriptCommitJson(data: unknown): ValidationError | undefined {
|
||||
const stack: Array<{ depth: number; value: unknown }> = [{ depth: 0, value: data }];
|
||||
const seen = new WeakSet<object>();
|
||||
while (stack.length > 0) {
|
||||
const current = stack.pop();
|
||||
if (!current) {
|
||||
break;
|
||||
}
|
||||
if (current.depth > WORKER_TRANSCRIPT_MAX_JSON_DEPTH) {
|
||||
return {
|
||||
keyword: "maxDepth",
|
||||
params: { limit: WORKER_TRANSCRIPT_MAX_JSON_DEPTH },
|
||||
message: `must not exceed JSON nesting depth ${WORKER_TRANSCRIPT_MAX_JSON_DEPTH}`,
|
||||
};
|
||||
}
|
||||
if (
|
||||
current.value === null ||
|
||||
typeof current.value === "string" ||
|
||||
typeof current.value === "boolean"
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
if (typeof current.value === "number") {
|
||||
if (!Number.isFinite(current.value)) {
|
||||
return { keyword: "finite", message: "must contain only finite JSON numbers" };
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (typeof current.value !== "object") {
|
||||
return { keyword: "jsonValue", message: "must contain only JSON values" };
|
||||
}
|
||||
if (seen.has(current.value)) {
|
||||
return { keyword: "acyclic", message: "must be an acyclic JSON value" };
|
||||
}
|
||||
seen.add(current.value);
|
||||
const values = Array.isArray(current.value)
|
||||
? current.value
|
||||
: Object.values(current.value as Record<string, unknown>);
|
||||
for (const value of values) {
|
||||
stack.push({ depth: current.depth + 1, value });
|
||||
}
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
export const validateWorkerTranscriptCommitParams = lazyCompile<WorkerTranscriptCommitParams>(
|
||||
WorkerTranscriptCommitParamsSchema,
|
||||
checkWorkerTranscriptCommitJson,
|
||||
);
|
||||
export const validateGatewaySuspendPrepareParams = lazyCompile<GatewaySuspendPrepareParams>(
|
||||
GatewaySuspendPrepareParamsSchema,
|
||||
);
|
||||
@@ -1517,6 +1593,13 @@ export {
|
||||
WorkerHeartbeatRequestFrameSchema,
|
||||
WorkerHeartbeatResponseFrameSchema,
|
||||
WorkerProtocolCloseReasonSchema,
|
||||
WorkerTranscriptCommitErrorReasonSchema,
|
||||
WorkerTranscriptCommitErrorShapeSchema,
|
||||
WorkerTranscriptCommitParamsSchema,
|
||||
WorkerTranscriptCommitRequestFrameSchema,
|
||||
WorkerTranscriptCommitResponseFrameSchema,
|
||||
WorkerTranscriptCommitResultSchema,
|
||||
WorkerTranscriptMessageSchema,
|
||||
WORKER_HEARTBEAT_INTERVAL_MS,
|
||||
WORKER_PROTOCOL_FEATURES,
|
||||
WORKER_PROTOCOL_MAX_FEATURE_LENGTH,
|
||||
@@ -1527,6 +1610,10 @@ export {
|
||||
WORKER_PROTOCOL_MAX_PAYLOAD_BYTES,
|
||||
WORKER_PROTOCOL_METHODS,
|
||||
WORKER_RPC_SET_VERSION,
|
||||
WORKER_TRANSCRIPT_MAX_BATCH_MESSAGES,
|
||||
WORKER_TRANSCRIPT_MAX_CONTENT_PARTS,
|
||||
WORKER_TRANSCRIPT_MAX_JSON_DEPTH,
|
||||
WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE,
|
||||
EnvironmentStatusSchema,
|
||||
WorkerEnvironmentStateSchema,
|
||||
WorkerTunnelStatusSchema,
|
||||
@@ -1902,6 +1989,13 @@ export type {
|
||||
WorkerHeartbeatResponseFrame,
|
||||
WorkerHelloOk,
|
||||
WorkerProtocolCloseReason,
|
||||
WorkerTranscriptCommitErrorReason,
|
||||
WorkerTranscriptCommitErrorShape,
|
||||
WorkerTranscriptCommitParams,
|
||||
WorkerTranscriptCommitRequestFrame,
|
||||
WorkerTranscriptCommitResponseFrame,
|
||||
WorkerTranscriptCommitResult,
|
||||
WorkerTranscriptMessage,
|
||||
GatewaySuspendTaskBlocker,
|
||||
GatewaySuspendBlocker,
|
||||
GatewaySuspendPrepareParams,
|
||||
|
||||
@@ -7,10 +7,14 @@ import {
|
||||
WorkerHeartbeatRequestFrameSchema,
|
||||
WorkerHeartbeatResponseFrameSchema,
|
||||
WorkerProtocolCloseReasonSchema,
|
||||
WorkerTranscriptCommitRequestFrameSchema,
|
||||
WorkerTranscriptCommitResponseFrameSchema,
|
||||
WORKER_RPC_SET_VERSION,
|
||||
WORKER_TRANSCRIPT_MAX_JSON_DEPTH,
|
||||
validateWorkerAdmissionHandshake,
|
||||
validateWorkerConnectRequestFrame,
|
||||
validateWorkerHeartbeatParams,
|
||||
validateWorkerTranscriptCommitParams,
|
||||
} from "../index.js";
|
||||
|
||||
const bundleHash = "a".repeat(64);
|
||||
@@ -49,6 +53,46 @@ const workerHello = {
|
||||
credentialExpiresAtMs: 10_000,
|
||||
policy: { heartbeatIntervalMs: 15_000, maxPayload: 1_024 },
|
||||
};
|
||||
const usage = {
|
||||
input: 1,
|
||||
output: 2,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 3,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
};
|
||||
const transcriptMessages = [
|
||||
{
|
||||
role: "user" as const,
|
||||
content: [{ type: "text" as const, text: "run the probe" }],
|
||||
timestamp: 1,
|
||||
},
|
||||
{
|
||||
role: "assistant" as const,
|
||||
content: [
|
||||
{
|
||||
type: "toolCall" as const,
|
||||
id: "call-1",
|
||||
name: "probe",
|
||||
arguments: { value: 1 },
|
||||
},
|
||||
],
|
||||
api: "responses",
|
||||
provider: "fixture",
|
||||
model: "fixture-model",
|
||||
usage,
|
||||
stopReason: "toolUse" as const,
|
||||
timestamp: 2,
|
||||
},
|
||||
{
|
||||
role: "toolResult" as const,
|
||||
toolCallId: "call-1",
|
||||
toolName: "probe",
|
||||
content: [{ type: "text" as const, text: "ok" }],
|
||||
isError: false,
|
||||
timestamp: 3,
|
||||
},
|
||||
];
|
||||
|
||||
describe("worker admission handshake schema", () => {
|
||||
it("accepts the bootstrap receipt and future unique feature names", () => {
|
||||
@@ -112,6 +156,139 @@ describe("worker protocol schemas", () => {
|
||||
expect(Value.Check(WorkerHeartbeatResponseFrameSchema, response)).toBe(true);
|
||||
});
|
||||
|
||||
it("accepts semantic transcript commits and generated-id responses", () => {
|
||||
const params = {
|
||||
runEpoch: 2,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: transcriptMessages,
|
||||
};
|
||||
expect(validateWorkerTranscriptCommitParams(params)).toBe(true);
|
||||
expect(
|
||||
Value.Check(WorkerTranscriptCommitRequestFrameSchema, {
|
||||
type: "req",
|
||||
id: "commit-1",
|
||||
method: "worker.transcript.commit",
|
||||
params,
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(
|
||||
Value.Check(WorkerTranscriptCommitResponseFrameSchema, {
|
||||
type: "res",
|
||||
id: "commit-1",
|
||||
ok: true,
|
||||
payload: { entryIds: ["entry-1", "entry-2", "entry-3"], newLeafId: "entry-3" },
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(
|
||||
Value.Check(WorkerTranscriptCommitResponseFrameSchema, {
|
||||
type: "res",
|
||||
id: "commit-1",
|
||||
ok: false,
|
||||
error: {
|
||||
code: "INVALID_REQUEST",
|
||||
message: "worker request rejected",
|
||||
details: { reason: "credential-replaced" },
|
||||
},
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(
|
||||
Value.Check(WorkerTranscriptCommitResponseFrameSchema, {
|
||||
type: "res",
|
||||
id: "commit-1",
|
||||
ok: false,
|
||||
error: {
|
||||
code: "INVALID_REQUEST",
|
||||
message: "transcript commit rejected",
|
||||
details: { reason: "stale-base-leaf" },
|
||||
},
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ runEpoch: 2, seq: 1, baseLeafId: null, messages: [] },
|
||||
{ runEpoch: 2, seq: 0, baseLeafId: null, messages: transcriptMessages },
|
||||
{ runEpoch: 2, seq: 1, baseLeafId: null, messages: transcriptMessages, sessionId: "other" },
|
||||
{
|
||||
runEpoch: 2,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: [{ ...transcriptMessages[0], id: "entry-from-worker" }],
|
||||
},
|
||||
{
|
||||
runEpoch: 2,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: [{ ...transcriptMessages[0], parentId: "parent-from-worker" }],
|
||||
},
|
||||
{
|
||||
runEpoch: 2,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: [{ ...transcriptMessages[0], sessionId: "foreign-session" }],
|
||||
},
|
||||
])("rejects raw transcript identity or invalid batch fields %#", (candidate) => {
|
||||
expect(validateWorkerTranscriptCommitParams(candidate)).toBe(false);
|
||||
});
|
||||
|
||||
it("rejects deeply nested worker JSON before schema compilation", () => {
|
||||
let nested: unknown = "leaf";
|
||||
for (let depth = 0; depth <= WORKER_TRANSCRIPT_MAX_JSON_DEPTH; depth += 1) {
|
||||
nested = { nested };
|
||||
}
|
||||
const assistant = transcriptMessages[1];
|
||||
if (!assistant || assistant.role !== "assistant") {
|
||||
throw new Error("expected assistant transcript fixture");
|
||||
}
|
||||
const candidate = {
|
||||
runEpoch: 2,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: [
|
||||
{
|
||||
...assistant,
|
||||
content: [
|
||||
{
|
||||
type: "toolCall" as const,
|
||||
id: "call-deep",
|
||||
name: "probe",
|
||||
arguments: { nested },
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
expect(validateWorkerTranscriptCommitParams(candidate)).toBe(false);
|
||||
expect(validateWorkerTranscriptCommitParams.errors?.[0]).toMatchObject({
|
||||
keyword: "maxDepth",
|
||||
params: { limit: WORKER_TRANSCRIPT_MAX_JSON_DEPTH },
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects non-finite numbers parsed from worker JSON", () => {
|
||||
const candidate = JSON.parse(`{
|
||||
"runEpoch": 2,
|
||||
"seq": 1,
|
||||
"baseLeafId": null,
|
||||
"messages": [{
|
||||
"role": "toolResult",
|
||||
"toolCallId": "call-non-finite",
|
||||
"toolName": "probe",
|
||||
"content": [],
|
||||
"details": { "value": 1e400 },
|
||||
"isError": false,
|
||||
"timestamp": 1
|
||||
}]
|
||||
}`) as unknown;
|
||||
|
||||
expect(validateWorkerTranscriptCommitParams(candidate)).toBe(false);
|
||||
expect(validateWorkerTranscriptCommitParams.errors?.[0]).toMatchObject({
|
||||
keyword: "finite",
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps worker close reasons closed", () => {
|
||||
expect(Value.Check(WorkerProtocolCloseReasonSchema, "credential-replaced")).toBe(true);
|
||||
expect(Value.Check(WorkerProtocolCloseReasonSchema, "not-a-worker-reason")).toBe(false);
|
||||
|
||||
@@ -1,16 +1,24 @@
|
||||
import { Type, type Static } from "typebox";
|
||||
import { GATEWAY_CLIENT_IDS, GATEWAY_CLIENT_MODES } from "../client-info.js";
|
||||
|
||||
// Additive RPCs require exact build-bound features; bump only for an incompatible base set.
|
||||
export const WORKER_RPC_SET_VERSION = 1;
|
||||
export const WORKER_HEARTBEAT_INTERVAL_MS = 15_000;
|
||||
export const WORKER_PROTOCOL_METHODS = ["worker.heartbeat"] as const;
|
||||
export const WORKER_PROTOCOL_FEATURES = ["worker-heartbeat-v1"] as const;
|
||||
export const WORKER_PROTOCOL_METHODS = ["worker.heartbeat", "worker.transcript.commit"] as const;
|
||||
export const WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE = "worker-transcript-commit-v1";
|
||||
export const WORKER_PROTOCOL_FEATURES = [
|
||||
"worker-heartbeat-v1",
|
||||
WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE,
|
||||
] as const;
|
||||
export const WORKER_PROTOCOL_MAX_IDENTIFIER_LENGTH = 256;
|
||||
export const WORKER_PROTOCOL_MAX_FRAME_ID_LENGTH = 128;
|
||||
export const WORKER_PROTOCOL_MAX_METHOD_LENGTH = 64;
|
||||
export const WORKER_PROTOCOL_MAX_PAYLOAD_BYTES = 64 * 1024;
|
||||
export const WORKER_PROTOCOL_MAX_FEATURES = 64;
|
||||
export const WORKER_PROTOCOL_MAX_FEATURE_LENGTH = 128;
|
||||
export const WORKER_TRANSCRIPT_MAX_BATCH_MESSAGES = 64;
|
||||
export const WORKER_TRANSCRIPT_MAX_CONTENT_PARTS = 128;
|
||||
export const WORKER_TRANSCRIPT_MAX_JSON_DEPTH = 32;
|
||||
|
||||
const WorkerIdentifierSchema = Type.String({
|
||||
minLength: 1,
|
||||
@@ -229,6 +237,262 @@ export const WorkerHeartbeatResponseFrameSchema = Type.Union([
|
||||
WorkerErrorResponseFrameSchema,
|
||||
]);
|
||||
|
||||
const WorkerTranscriptTextContentSchema = Type.Object(
|
||||
{
|
||||
type: Type.Literal("text"),
|
||||
text: Type.String({ maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES }),
|
||||
textSignature: Type.Optional(
|
||||
Type.String({ minLength: 1, maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES }),
|
||||
),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptThinkingContentSchema = Type.Object(
|
||||
{
|
||||
type: Type.Literal("thinking"),
|
||||
thinking: Type.String({ maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES }),
|
||||
thinkingSignature: Type.Optional(
|
||||
Type.String({ minLength: 1, maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES }),
|
||||
),
|
||||
redacted: Type.Optional(Type.Boolean()),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptImageContentSchema = Type.Object(
|
||||
{
|
||||
type: Type.Literal("image"),
|
||||
data: Type.String({ minLength: 1, maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES }),
|
||||
mimeType: Type.String({ minLength: 1, maxLength: 256 }),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptToolCallSchema = Type.Object(
|
||||
{
|
||||
type: Type.Literal("toolCall"),
|
||||
id: WorkerIdentifierSchema,
|
||||
name: WorkerIdentifierSchema,
|
||||
arguments: Type.Record(Type.String({ minLength: 1, maxLength: 256 }), Type.Unknown()),
|
||||
thoughtSignature: Type.Optional(
|
||||
Type.String({ minLength: 1, maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES }),
|
||||
),
|
||||
executionMode: Type.Optional(
|
||||
Type.Union([Type.Literal("sequential"), Type.Literal("parallel")]),
|
||||
),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptUsageSchema = Type.Object(
|
||||
{
|
||||
input: Type.Number({ minimum: 0 }),
|
||||
output: Type.Number({ minimum: 0 }),
|
||||
cacheRead: Type.Number({ minimum: 0 }),
|
||||
cacheWrite: Type.Number({ minimum: 0 }),
|
||||
contextUsage: Type.Optional(
|
||||
Type.Union([
|
||||
Type.Object(
|
||||
{
|
||||
state: Type.Literal("available"),
|
||||
promptTokens: Type.Number({ minimum: 0 }),
|
||||
totalTokens: Type.Number({ minimum: 0 }),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
),
|
||||
Type.Object({ state: Type.Literal("unavailable") }, { additionalProperties: false }),
|
||||
]),
|
||||
),
|
||||
totalTokens: Type.Number({ minimum: 0 }),
|
||||
cost: Type.Object(
|
||||
{
|
||||
input: Type.Number({ minimum: 0 }),
|
||||
output: Type.Number({ minimum: 0 }),
|
||||
cacheRead: Type.Number({ minimum: 0 }),
|
||||
cacheWrite: Type.Number({ minimum: 0 }),
|
||||
total: Type.Number({ minimum: 0 }),
|
||||
totalOrigin: Type.Optional(Type.Literal("provider-billed")),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptAssistantDiagnosticSchema = Type.Object(
|
||||
{
|
||||
type: WorkerIdentifierSchema,
|
||||
timestamp: Type.Integer({ minimum: 0 }),
|
||||
error: Type.Optional(
|
||||
Type.Object(
|
||||
{
|
||||
name: Type.Optional(Type.String({ maxLength: 256 })),
|
||||
message: Type.String({ maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES }),
|
||||
stack: Type.Optional(Type.String({ maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES })),
|
||||
code: Type.Optional(Type.Union([Type.String({ maxLength: 256 }), Type.Number()])),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
),
|
||||
),
|
||||
details: Type.Optional(
|
||||
Type.Record(Type.String({ minLength: 1, maxLength: 256 }), Type.Unknown()),
|
||||
),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptUserMessageSchema = Type.Object(
|
||||
{
|
||||
role: Type.Literal("user"),
|
||||
content: Type.Array(
|
||||
Type.Union([WorkerTranscriptTextContentSchema, WorkerTranscriptImageContentSchema]),
|
||||
{ minItems: 1, maxItems: WORKER_TRANSCRIPT_MAX_CONTENT_PARTS },
|
||||
),
|
||||
timestamp: Type.Integer({ minimum: 0 }),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptAssistantMessageSchema = Type.Object(
|
||||
{
|
||||
role: Type.Literal("assistant"),
|
||||
content: Type.Array(
|
||||
Type.Union([
|
||||
WorkerTranscriptTextContentSchema,
|
||||
WorkerTranscriptThinkingContentSchema,
|
||||
WorkerTranscriptToolCallSchema,
|
||||
]),
|
||||
{ maxItems: WORKER_TRANSCRIPT_MAX_CONTENT_PARTS },
|
||||
),
|
||||
api: WorkerIdentifierSchema,
|
||||
provider: WorkerIdentifierSchema,
|
||||
model: WorkerIdentifierSchema,
|
||||
responseModel: Type.Optional(WorkerIdentifierSchema),
|
||||
responseId: Type.Optional(WorkerIdentifierSchema),
|
||||
diagnostics: Type.Optional(
|
||||
Type.Array(WorkerTranscriptAssistantDiagnosticSchema, {
|
||||
maxItems: WORKER_TRANSCRIPT_MAX_CONTENT_PARTS,
|
||||
}),
|
||||
),
|
||||
usage: WorkerTranscriptUsageSchema,
|
||||
stopReason: Type.Union([
|
||||
Type.Literal("stop"),
|
||||
Type.Literal("length"),
|
||||
Type.Literal("toolUse"),
|
||||
Type.Literal("error"),
|
||||
Type.Literal("aborted"),
|
||||
]),
|
||||
errorMessage: Type.Optional(Type.String({ maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES })),
|
||||
errorCode: Type.Optional(Type.String({ maxLength: 256 })),
|
||||
errorType: Type.Optional(Type.String({ maxLength: 256 })),
|
||||
errorBody: Type.Optional(Type.String({ maxLength: WORKER_PROTOCOL_MAX_PAYLOAD_BYTES })),
|
||||
timestamp: Type.Integer({ minimum: 0 }),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptToolResultMessageSchema = Type.Object(
|
||||
{
|
||||
role: Type.Literal("toolResult"),
|
||||
toolCallId: WorkerIdentifierSchema,
|
||||
toolName: WorkerIdentifierSchema,
|
||||
content: Type.Array(
|
||||
Type.Union([WorkerTranscriptTextContentSchema, WorkerTranscriptImageContentSchema]),
|
||||
{ maxItems: WORKER_TRANSCRIPT_MAX_CONTENT_PARTS },
|
||||
),
|
||||
details: Type.Optional(Type.Unknown()),
|
||||
isError: Type.Boolean(),
|
||||
timestamp: Type.Integer({ minimum: 0 }),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
export const WorkerTranscriptMessageSchema = Type.Union([
|
||||
WorkerTranscriptUserMessageSchema,
|
||||
WorkerTranscriptAssistantMessageSchema,
|
||||
WorkerTranscriptToolResultMessageSchema,
|
||||
]);
|
||||
|
||||
export const WorkerTranscriptCommitParamsSchema = Type.Object(
|
||||
{
|
||||
runEpoch: Type.Integer({ minimum: 0, maximum: Number.MAX_SAFE_INTEGER }),
|
||||
seq: Type.Integer({ minimum: 1, maximum: Number.MAX_SAFE_INTEGER }),
|
||||
baseLeafId: Type.Union([WorkerIdentifierSchema, Type.Null()]),
|
||||
messages: Type.Array(WorkerTranscriptMessageSchema, {
|
||||
minItems: 1,
|
||||
maxItems: WORKER_TRANSCRIPT_MAX_BATCH_MESSAGES,
|
||||
}),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
export const WorkerTranscriptCommitResultSchema = Type.Object(
|
||||
{
|
||||
entryIds: Type.Array(WorkerIdentifierSchema, {
|
||||
minItems: 1,
|
||||
maxItems: WORKER_TRANSCRIPT_MAX_BATCH_MESSAGES,
|
||||
}),
|
||||
newLeafId: WorkerIdentifierSchema,
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
export const WorkerTranscriptCommitErrorReasonSchema = Type.Union([
|
||||
Type.Literal("stale-base-leaf"),
|
||||
Type.Literal("epoch-mismatch"),
|
||||
Type.Literal("invalid-batch"),
|
||||
Type.Literal("session-not-attached"),
|
||||
]);
|
||||
|
||||
export const WorkerTranscriptCommitErrorShapeSchema = Type.Object(
|
||||
{
|
||||
code: Type.Literal("INVALID_REQUEST"),
|
||||
message: Type.String({ minLength: 1, maxLength: 256 }),
|
||||
details: Type.Object(
|
||||
{ reason: WorkerTranscriptCommitErrorReasonSchema },
|
||||
{ additionalProperties: false },
|
||||
),
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
export const WorkerTranscriptCommitRequestFrameSchema = Type.Object(
|
||||
{
|
||||
type: Type.Literal("req"),
|
||||
id: WorkerFrameIdSchema,
|
||||
method: Type.Literal(WORKER_PROTOCOL_METHODS[1]),
|
||||
params: WorkerTranscriptCommitParamsSchema,
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptCommitSuccessResponseFrameSchema = Type.Object(
|
||||
{
|
||||
type: Type.Literal("res"),
|
||||
id: WorkerFrameIdSchema,
|
||||
ok: Type.Literal(true),
|
||||
payload: WorkerTranscriptCommitResultSchema,
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
const WorkerTranscriptCommitErrorResponseFrameSchema = Type.Object(
|
||||
{
|
||||
type: Type.Literal("res"),
|
||||
id: WorkerFrameIdSchema,
|
||||
ok: Type.Literal(false),
|
||||
error: WorkerTranscriptCommitErrorShapeSchema,
|
||||
},
|
||||
{ additionalProperties: false },
|
||||
);
|
||||
|
||||
export const WorkerTranscriptCommitResponseFrameSchema = Type.Union([
|
||||
WorkerTranscriptCommitSuccessResponseFrameSchema,
|
||||
WorkerTranscriptCommitErrorResponseFrameSchema,
|
||||
WorkerErrorResponseFrameSchema,
|
||||
]);
|
||||
|
||||
export type WorkerAdmissionHandshake = Static<typeof WorkerAdmissionHandshakeSchema>;
|
||||
export type WorkerConnectParams = Static<typeof WorkerConnectParamsSchema>;
|
||||
export type WorkerConnectRequestFrame = Static<typeof WorkerConnectRequestFrameSchema>;
|
||||
@@ -241,3 +505,18 @@ export type WorkerHeartbeatParams = Static<typeof WorkerHeartbeatParamsSchema>;
|
||||
export type WorkerHeartbeatResult = Static<typeof WorkerHeartbeatResultSchema>;
|
||||
export type WorkerHeartbeatRequestFrame = Static<typeof WorkerHeartbeatRequestFrameSchema>;
|
||||
export type WorkerHeartbeatResponseFrame = Static<typeof WorkerHeartbeatResponseFrameSchema>;
|
||||
export type WorkerTranscriptMessage = Static<typeof WorkerTranscriptMessageSchema>;
|
||||
export type WorkerTranscriptCommitParams = Static<typeof WorkerTranscriptCommitParamsSchema>;
|
||||
export type WorkerTranscriptCommitResult = Static<typeof WorkerTranscriptCommitResultSchema>;
|
||||
export type WorkerTranscriptCommitErrorReason = Static<
|
||||
typeof WorkerTranscriptCommitErrorReasonSchema
|
||||
>;
|
||||
export type WorkerTranscriptCommitErrorShape = Static<
|
||||
typeof WorkerTranscriptCommitErrorShapeSchema
|
||||
>;
|
||||
export type WorkerTranscriptCommitRequestFrame = Static<
|
||||
typeof WorkerTranscriptCommitRequestFrameSchema
|
||||
>;
|
||||
export type WorkerTranscriptCommitResponseFrame = Static<
|
||||
typeof WorkerTranscriptCommitResponseFrameSchema
|
||||
>;
|
||||
|
||||
@@ -51,6 +51,7 @@ import {
|
||||
publishOwnedSessionFileSnapshot,
|
||||
} from "../../config/sessions/transcript-write-context.js";
|
||||
import { CURRENT_SESSION_VERSION } from "../../config/sessions/version.js";
|
||||
import type { OpenClawConfig } from "../../config/types.openclaw.js";
|
||||
import type { ImageContent, Message, TextContent } from "../../llm/types.js";
|
||||
import { logWarn } from "../../logger.js";
|
||||
import { getAgentDir as getDefaultAgentDir, getSessionsDir } from "../config.js";
|
||||
@@ -216,6 +217,8 @@ export type SessionEntry =
|
||||
export type FileEntry = SessionHeader | SessionEntry;
|
||||
|
||||
type AppendPersistenceOptions = {
|
||||
config?: OpenClawConfig;
|
||||
idempotencyLookup?: "scan" | "scan-assistant" | "caller-checked";
|
||||
invalidateSerializedPrefixCache?: boolean;
|
||||
};
|
||||
|
||||
@@ -2245,7 +2248,7 @@ export class SessionManager {
|
||||
publishSnapshot = true,
|
||||
): void {
|
||||
if (this.sqlitePersistence) {
|
||||
this.persistSqliteRecord(entry);
|
||||
this.persistSqliteRecord(entry, options);
|
||||
return;
|
||||
}
|
||||
if (!this.shouldPersist || !this.sessionFile) {
|
||||
@@ -2319,7 +2322,7 @@ export class SessionManager {
|
||||
this.persistRecord(entry, options);
|
||||
}
|
||||
|
||||
private persistSqliteRecord(entry: unknown): void {
|
||||
private persistSqliteRecord(entry: unknown, options?: AppendPersistenceOptions): void {
|
||||
if (!isIndexedSessionEntry(entry)) {
|
||||
return;
|
||||
}
|
||||
@@ -2337,13 +2340,21 @@ export class SessionManager {
|
||||
appendTranscriptEventSync(scope, entry);
|
||||
return;
|
||||
}
|
||||
appendTranscriptMessageSync(scope, {
|
||||
const result = appendTranscriptMessageSync(scope, {
|
||||
cwd: this.cwd,
|
||||
eventId: entry.id,
|
||||
...(options?.config ? { config: options.config } : {}),
|
||||
...(options?.idempotencyLookup ? { idempotencyLookup: options.idempotencyLookup } : {}),
|
||||
message: entry.message,
|
||||
now: Date.parse(entry.timestamp),
|
||||
parentId: entry.parentId,
|
||||
});
|
||||
if (
|
||||
options?.idempotencyLookup === "caller-checked" &&
|
||||
(!result?.appended || result.messageId !== entry.id)
|
||||
) {
|
||||
throw new Error(`Session transcript append was not persisted: ${entry.id}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1943,6 +1943,21 @@ export async function withSqliteTranscriptWriteLock<T>(
|
||||
});
|
||||
}
|
||||
|
||||
/** Runs synchronous transcript work under one writer queue and SQLite transaction. */
|
||||
export async function withSqliteTranscriptWriteTransaction<T>(
|
||||
scope: SessionTranscriptWriteScope,
|
||||
run: (context: { sessionFile: string }) => T,
|
||||
): Promise<T> {
|
||||
const resolved = resolveSqliteTranscriptScope(scope);
|
||||
return await runExclusiveSqliteSessionWrite(resolved, async () =>
|
||||
runOpenClawAgentWriteTransaction(
|
||||
() => run({ sessionFile: formatSqliteSessionMarkerForScope(resolved) }),
|
||||
toDatabaseOptions(resolved),
|
||||
{ operationLabel: "session.transcript.batch" },
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
function appendSqliteTranscriptMessageInTransaction<TMessage>(
|
||||
database: OpenClawAgentDatabase,
|
||||
resolved: ResolvedTranscriptScope,
|
||||
|
||||
@@ -71,6 +71,7 @@ import {
|
||||
updateSqliteSessionEntry,
|
||||
upsertSqliteSessionEntry,
|
||||
withSqliteTranscriptWriteLock,
|
||||
withSqliteTranscriptWriteTransaction,
|
||||
} from "./session-accessor.sqlite.js";
|
||||
import {
|
||||
formatSqliteSessionFileMarker,
|
||||
@@ -349,6 +350,11 @@ export type SessionTranscriptWriteLockAccessorContext = {
|
||||
replaceEvents: (events: readonly TranscriptEvent[]) => Promise<void>;
|
||||
};
|
||||
|
||||
export type SessionTranscriptWriteTransactionContext = {
|
||||
/** Canonical marker for the same agent database owned by the transaction. */
|
||||
sessionFile: string;
|
||||
};
|
||||
|
||||
export type SessionTranscriptTurnUpdateMode = "inline" | "file-only" | "none";
|
||||
|
||||
export type SessionTranscriptTurnMessageAppend = TranscriptMessageAppendOptions<unknown> & {
|
||||
@@ -2435,6 +2441,14 @@ export async function withTranscriptWriteLock<T>(
|
||||
return await withSqliteTranscriptWriteLock(scope, run);
|
||||
}
|
||||
|
||||
/** Runs a synchronous DAG batch under one transcript writer queue and transaction. */
|
||||
export async function withTranscriptWriteTransaction<T>(
|
||||
scope: SessionTranscriptWriteScope,
|
||||
run: (context: SessionTranscriptWriteTransactionContext) => T,
|
||||
): Promise<T> {
|
||||
return await withSqliteTranscriptWriteTransaction(scope, run);
|
||||
}
|
||||
|
||||
/**
|
||||
* Trims a transcript for manual sessions.compact and clears stale token metadata.
|
||||
* This is one storage-sized mutation: future stores can trim transcript rows and
|
||||
|
||||
@@ -138,6 +138,7 @@ import { maybeSeedControlUiAllowedOriginsAtStartup } from "./startup-control-ui-
|
||||
import type { WorkerBundleProducer, WorkerNpmArtifact } from "./worker-environments/bundle.js";
|
||||
import { createWorkerEnvironmentService } from "./worker-environments/service.js";
|
||||
import { createWorkerEnvironmentStore } from "./worker-environments/store.js";
|
||||
import { createWorkerTranscriptCommitter } from "./worker-environments/transcript-commit.js";
|
||||
|
||||
type LoadGatewayModelCatalog = typeof import("./server-model-catalog.js").loadGatewayModelCatalog;
|
||||
type LoadGatewayModelCatalogSnapshot =
|
||||
@@ -795,6 +796,9 @@ export async function startGatewayServer(
|
||||
prepareInstallation: prepareWorkerInstallation,
|
||||
tunnelManager: workerTunnelManager,
|
||||
resolveWorkerGateway: () => resolveWorkerGatewayEndpoint(),
|
||||
applyTranscriptCommit: createWorkerTranscriptCommitter({
|
||||
getConfig: getRuntimeConfig,
|
||||
}).commit,
|
||||
resolveSshIdentity: async ({ provider, leaseId, profile, keyRef }) => {
|
||||
const workerEnvironmentRuntime = await loadWorkerEnvironmentRuntimeModule();
|
||||
return await workerEnvironmentRuntime.resolveWorkerSshIdentity({
|
||||
|
||||
@@ -8,6 +8,8 @@ import {
|
||||
PROTOCOL_VERSION,
|
||||
type WorkerAdmissionFailureReason,
|
||||
type WorkerConnectParams,
|
||||
type WorkerTranscriptCommitErrorReason,
|
||||
WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE,
|
||||
} from "../../../../packages/gateway-protocol/src/index.js";
|
||||
import {
|
||||
resetGatewayWorkAdmission,
|
||||
@@ -22,7 +24,7 @@ const CREDENTIAL = ["worker", "credential", "fixture"].join("-");
|
||||
const HANDSHAKE = {
|
||||
bundleHash: "a".repeat(64),
|
||||
openclawVersion: "2026.7.11",
|
||||
protocolFeatures: ["worker-heartbeat-v1"],
|
||||
protocolFeatures: ["worker-heartbeat-v1", WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE],
|
||||
};
|
||||
const WORKER_CONNECT: WorkerConnectParams = {
|
||||
minProtocol: PROTOCOL_VERSION,
|
||||
@@ -53,6 +55,18 @@ const IDENTITY: WorkerConnectionIdentity = {
|
||||
protocolFeatures: [...HANDSHAKE.protocolFeatures],
|
||||
credentialExpiresAtMs: Date.now() + 60_000,
|
||||
};
|
||||
const TRANSCRIPT_COMMIT = {
|
||||
runEpoch: 1,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: [
|
||||
{
|
||||
role: "user" as const,
|
||||
content: [{ type: "text" as const, text: "hello" }],
|
||||
timestamp: 1,
|
||||
},
|
||||
],
|
||||
};
|
||||
const cleanups: Array<() => void> = [];
|
||||
|
||||
function createLogger() {
|
||||
@@ -62,6 +76,8 @@ function createLogger() {
|
||||
function attachHarness(
|
||||
options: {
|
||||
admissionFailure?: WorkerAdmissionFailureReason;
|
||||
commitFailure?: WorkerTranscriptCommitErrorReason;
|
||||
identity?: WorkerConnectionIdentity;
|
||||
validationFailure?: ReturnType<WorkerConnectionService["validateWorkerConnection"]>;
|
||||
} = {},
|
||||
) {
|
||||
@@ -72,7 +88,15 @@ function attachHarness(
|
||||
admitWorker: vi.fn(async () =>
|
||||
options.admissionFailure
|
||||
? { ok: false as const, reason: options.admissionFailure }
|
||||
: { ok: true as const, identity: IDENTITY },
|
||||
: { ok: true as const, identity: options.identity ?? IDENTITY },
|
||||
),
|
||||
commitTranscript: vi.fn(async () =>
|
||||
options.commitFailure
|
||||
? { ok: false as const, reason: options.commitFailure }
|
||||
: {
|
||||
ok: true as const,
|
||||
result: { entryIds: ["entry-1"], newLeafId: "entry-1" },
|
||||
},
|
||||
),
|
||||
validateWorkerConnection: vi.fn(() => options.validationFailure ?? null),
|
||||
} as WorkerConnectionService;
|
||||
@@ -83,6 +107,7 @@ function attachHarness(
|
||||
});
|
||||
const logGateway = createLogger();
|
||||
const logWsControl = createLogger();
|
||||
const setLastFrameMeta = vi.fn();
|
||||
const cleanup = attachWorkerWsMessageHandler({
|
||||
socket: socket as unknown as WebSocket,
|
||||
connId: "worker-connection",
|
||||
@@ -96,7 +121,7 @@ function attachHarness(
|
||||
setHandshakeState: vi.fn(),
|
||||
advanceHandshakePhase: vi.fn(),
|
||||
setCloseCause: vi.fn(),
|
||||
setLastFrameMeta: vi.fn(),
|
||||
setLastFrameMeta,
|
||||
logGateway,
|
||||
logWsControl,
|
||||
});
|
||||
@@ -110,6 +135,7 @@ function attachHarness(
|
||||
responses,
|
||||
service,
|
||||
setClient,
|
||||
setLastFrameMeta,
|
||||
sendRequest: (method: string, params: unknown) =>
|
||||
send({ type: "req", id: "request-1", method, params }),
|
||||
sendConnect: () =>
|
||||
@@ -181,6 +207,83 @@ describe("dedicated worker websocket protocol", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("dispatches semantic transcript commits on the closed worker allowlist", async () => {
|
||||
const harness = attachHarness();
|
||||
await admit(harness);
|
||||
harness.sendRequest("worker.transcript.commit", TRANSCRIPT_COMMIT);
|
||||
|
||||
await vi.waitFor(() => expect(harness.responses).toHaveLength(2));
|
||||
expect(harness.responses[1]).toMatchObject({
|
||||
ok: true,
|
||||
payload: { entryIds: ["entry-1"], newLeafId: "entry-1" },
|
||||
});
|
||||
expect(harness.service.commitTranscript).toHaveBeenCalledWith(IDENTITY, TRANSCRIPT_COMMIT);
|
||||
expect(harness.setLastFrameMeta).toHaveBeenLastCalledWith({
|
||||
type: "req",
|
||||
method: "worker.transcript.commit",
|
||||
});
|
||||
expect(harness.close).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects transcript commits when the admitted worker lacks the feature", async () => {
|
||||
const harness = attachHarness({
|
||||
identity: { ...IDENTITY, protocolFeatures: ["worker-heartbeat-v1"] },
|
||||
});
|
||||
await admit(harness);
|
||||
harness.sendRequest("worker.transcript.commit", TRANSCRIPT_COMMIT);
|
||||
|
||||
await vi.waitFor(() => expect(harness.close).toHaveBeenCalledWith(1008, "method-not-allowed"));
|
||||
expect(harness.service.commitTranscript).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns closed transcript errors without closing the worker connection", async () => {
|
||||
const harness = attachHarness({ commitFailure: "stale-base-leaf" });
|
||||
await admit(harness);
|
||||
harness.sendRequest("worker.transcript.commit", TRANSCRIPT_COMMIT);
|
||||
|
||||
await vi.waitFor(() => expect(harness.responses).toHaveLength(2));
|
||||
expect(harness.responses[1]).toMatchObject({
|
||||
ok: false,
|
||||
error: { details: { reason: "stale-base-leaf" } },
|
||||
});
|
||||
expect(harness.close).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects structurally invalid transcript batches before application", async () => {
|
||||
const harness = attachHarness();
|
||||
await admit(harness);
|
||||
harness.sendRequest("worker.transcript.commit", {
|
||||
...TRANSCRIPT_COMMIT,
|
||||
sessionId: "foreign-session",
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(harness.responses).toHaveLength(2));
|
||||
expect(harness.responses[1]).toMatchObject({
|
||||
ok: false,
|
||||
error: { details: { reason: "invalid-batch" } },
|
||||
});
|
||||
expect(harness.service.commitTranscript).not.toHaveBeenCalled();
|
||||
expect(harness.close).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("closes a replaced worker before parsing a malformed transcript batch", async () => {
|
||||
const harness = attachHarness();
|
||||
await admit(harness);
|
||||
vi.mocked(harness.service.validateWorkerConnection).mockReturnValue("credential-replaced");
|
||||
harness.sendRequest("worker.transcript.commit", {
|
||||
...TRANSCRIPT_COMMIT,
|
||||
sessionId: "foreign-session",
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(harness.responses).toHaveLength(2));
|
||||
expect(harness.responses[1]).toMatchObject({
|
||||
ok: false,
|
||||
error: { details: { reason: "credential-replaced" } },
|
||||
});
|
||||
await vi.waitFor(() => expect(harness.close).toHaveBeenCalledWith(1008, "credential-replaced"));
|
||||
expect(harness.service.commitTranscript).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("revalidates ownership immediately before admission", async () => {
|
||||
const harness = attachHarness({ validationFailure: "credential-replaced" });
|
||||
harness.sendConnect();
|
||||
|
||||
@@ -8,14 +8,18 @@ import {
|
||||
type WorkerHeartbeatResult,
|
||||
type WorkerHelloOk,
|
||||
type WorkerProtocolCloseReason,
|
||||
type WorkerTranscriptCommitErrorReason,
|
||||
type WorkerTranscriptCommitErrorShape,
|
||||
WORKER_HEARTBEAT_INTERVAL_MS,
|
||||
WORKER_PROTOCOL_MAX_FRAME_ID_LENGTH,
|
||||
WORKER_PROTOCOL_MAX_METHOD_LENGTH,
|
||||
WORKER_PROTOCOL_MAX_PAYLOAD_BYTES,
|
||||
WORKER_PROTOCOL_METHODS,
|
||||
WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE,
|
||||
validateRequestFrame,
|
||||
validateWorkerConnectRequestFrame,
|
||||
validateWorkerHeartbeatParams,
|
||||
validateWorkerTranscriptCommitParams,
|
||||
} from "../../../../packages/gateway-protocol/src/index.js";
|
||||
import { GATEWAY_STARTUP_RETRY_AFTER_MS } from "../../../../packages/gateway-protocol/src/startup-unavailable.js";
|
||||
import { rawDataToString } from "../../../infra/ws.js";
|
||||
@@ -26,10 +30,14 @@ import type { GatewayWsClient, WsHandshakePhase } from "../ws-types.js";
|
||||
|
||||
export type WorkerConnectionService = Pick<
|
||||
WorkerEnvironmentService,
|
||||
"admitWorker" | "validateWorkerConnection"
|
||||
"admitWorker" | "commitTranscript" | "validateWorkerConnection"
|
||||
>;
|
||||
|
||||
type WorkerRespond = (ok: boolean, payload?: unknown, error?: WorkerErrorShape) => void;
|
||||
type WorkerRespond = (
|
||||
ok: boolean,
|
||||
payload?: unknown,
|
||||
error?: WorkerErrorShape | WorkerTranscriptCommitErrorShape,
|
||||
) => void;
|
||||
type WorkerLogger = { warn(message: string): void };
|
||||
const MAX_QUEUED_WORKER_FRAMES = 16;
|
||||
|
||||
@@ -97,22 +105,56 @@ function rejectWorkerRequest(params: {
|
||||
queueMicrotask(() => params.close(1008, params.reason));
|
||||
}
|
||||
|
||||
function workerTranscriptCommitError(
|
||||
reason: WorkerTranscriptCommitErrorReason,
|
||||
): WorkerTranscriptCommitErrorShape {
|
||||
return {
|
||||
code: ErrorCodes.INVALID_REQUEST,
|
||||
message: "worker transcript commit rejected",
|
||||
details: { reason },
|
||||
};
|
||||
}
|
||||
|
||||
/** Closed worker dispatcher. It never calls the generic gateway method registry. */
|
||||
function dispatchWorkerRequest(params: {
|
||||
async function dispatchWorkerRequest(params: {
|
||||
request: RequestFrame;
|
||||
identity: WorkerConnectionIdentity;
|
||||
service: WorkerConnectionService | undefined;
|
||||
respond: WorkerRespond;
|
||||
close(code: number, reason: WorkerProtocolCloseReason): void;
|
||||
warn(message: string): void;
|
||||
}): void {
|
||||
const ownershipFailure = params.service
|
||||
? params.service.validateWorkerConnection(params.identity)
|
||||
: "environment-unavailable";
|
||||
}): Promise<void> {
|
||||
const service = params.service;
|
||||
if (!service) {
|
||||
rejectWorkerRequest({ ...params, reason: "environment-unavailable" });
|
||||
return;
|
||||
}
|
||||
const ownershipFailure = service.validateWorkerConnection(params.identity);
|
||||
if (ownershipFailure) {
|
||||
rejectWorkerRequest({ ...params, reason: ownershipFailure });
|
||||
return;
|
||||
}
|
||||
if (params.request.method === WORKER_PROTOCOL_METHODS[1]) {
|
||||
if (!params.identity.protocolFeatures.includes(WORKER_TRANSCRIPT_COMMIT_PROTOCOL_FEATURE)) {
|
||||
rejectWorkerRequest({ ...params, reason: "method-not-allowed" });
|
||||
return;
|
||||
}
|
||||
if (!validateWorkerTranscriptCommitParams(params.request.params)) {
|
||||
params.respond(false, undefined, workerTranscriptCommitError("invalid-batch"));
|
||||
return;
|
||||
}
|
||||
const outcome = await service.commitTranscript(params.identity, params.request.params);
|
||||
if (outcome.ok) {
|
||||
params.respond(true, outcome.result);
|
||||
return;
|
||||
}
|
||||
if ("closeReason" in outcome) {
|
||||
rejectWorkerRequest({ ...params, reason: outcome.closeReason });
|
||||
return;
|
||||
}
|
||||
params.respond(false, undefined, workerTranscriptCommitError(outcome.reason));
|
||||
return;
|
||||
}
|
||||
if (params.request.method !== WORKER_PROTOCOL_METHODS[0]) {
|
||||
rejectWorkerRequest({ ...params, reason: "method-not-allowed" });
|
||||
return;
|
||||
@@ -298,14 +340,17 @@ export function attachWorkerWsMessageHandler(params: WorkerWsMessageHandlerParam
|
||||
closeWorker(1008, "invalid-frame");
|
||||
return;
|
||||
}
|
||||
if (parsed.method === WORKER_PROTOCOL_METHODS[0]) {
|
||||
params.setLastFrameMeta({ type: "req", method: WORKER_PROTOCOL_METHODS[0] });
|
||||
if (
|
||||
parsed.method === WORKER_PROTOCOL_METHODS[0] ||
|
||||
parsed.method === WORKER_PROTOCOL_METHODS[1]
|
||||
) {
|
||||
params.setLastFrameMeta({ type: "req", method: parsed.method });
|
||||
}
|
||||
if (!client.worker) {
|
||||
closeWorker(1008, "environment-unavailable");
|
||||
return;
|
||||
}
|
||||
dispatchWorkerRequest({
|
||||
await dispatchWorkerRequest({
|
||||
request: parsed,
|
||||
identity: client.worker,
|
||||
service: params.service,
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
persistSessionTranscriptTurn,
|
||||
} from "../config/sessions/session-accessor.js";
|
||||
import { appendAssistantMessageToSessionTranscript } from "../config/sessions/transcript.js";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import { emitSessionLifecycleEvent } from "../sessions/session-lifecycle-events.js";
|
||||
import * as transcriptEvents from "../sessions/transcript-events.js";
|
||||
import {
|
||||
@@ -25,6 +26,9 @@ import {
|
||||
rpcReq,
|
||||
writeSessionStore,
|
||||
} from "./test-helpers.server.js";
|
||||
import type { WorkerConnectionIdentity } from "./worker-environments/connection-identity.js";
|
||||
import type { WorkerTranscriptCommitStore } from "./worker-environments/transcript-commit-store.js";
|
||||
import { createWorkerTranscriptCommitter } from "./worker-environments/transcript-commit.js";
|
||||
|
||||
installGatewayTestHooks({ scope: "suite" });
|
||||
|
||||
@@ -1210,6 +1214,97 @@ describe("session.message websocket events", () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("streams every worker transcript commit entry to the selected session subscriber", async () => {
|
||||
const storePath = await createSessionStoreFile();
|
||||
const sessionId = "sess-worker-commit-fanout";
|
||||
const sessionKey = "agent:main:worker";
|
||||
await writeSessionStore({
|
||||
entries: {
|
||||
worker: {
|
||||
sessionId,
|
||||
updatedAt: Date.now(),
|
||||
},
|
||||
},
|
||||
storePath,
|
||||
});
|
||||
const config: OpenClawConfig = {
|
||||
agents: { list: [{ id: "main", default: true }] },
|
||||
session: { mainKey: "main", store: storePath },
|
||||
};
|
||||
const ledger: WorkerTranscriptCommitStore = {
|
||||
begin: () => ({ kind: "claimed" }),
|
||||
complete: ({ outcome }) => outcome,
|
||||
};
|
||||
const committer = createWorkerTranscriptCommitter({ getConfig: () => config, store: ledger });
|
||||
const identity: WorkerConnectionIdentity = {
|
||||
environmentId: "environment-fanout",
|
||||
credentialHash: ["fanout", "credential", "hash"].join("-"),
|
||||
bundleHash: "f".repeat(64),
|
||||
sessionId,
|
||||
ownerEpoch: 4,
|
||||
rpcSetVersion: 1,
|
||||
protocolFeatures: ["worker-transcript-commit-v1"],
|
||||
credentialExpiresAtMs: Date.now() + 10_000,
|
||||
};
|
||||
|
||||
const ws = await harness.openWs();
|
||||
try {
|
||||
await connectOk(ws, { scopes: ["operator.read"] });
|
||||
const subscribeRes = await rpcReq(ws, "sessions.messages.subscribe", { key: sessionKey });
|
||||
expect(subscribeRes.ok).toBe(true);
|
||||
expect(subscribeRes.payload?.subscribed).toBe(true);
|
||||
expect(subscribeRes.payload?.key).toBe(sessionKey);
|
||||
|
||||
const eventPromises = [1, 2, 3].map((messageSeq) =>
|
||||
onceMessage(
|
||||
ws,
|
||||
(message) =>
|
||||
message.type === "event" &&
|
||||
message.event === "session.message" &&
|
||||
(message.payload as { sessionKey?: unknown } | undefined)?.sessionKey === sessionKey &&
|
||||
(message.payload as { messageSeq?: unknown } | undefined)?.messageSeq === messageSeq,
|
||||
),
|
||||
);
|
||||
const outcome = await committer.commit({
|
||||
identity,
|
||||
request: {
|
||||
runEpoch: identity.ownerEpoch,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: ["first", "second", "third"].map((text, index) => ({
|
||||
role: "user" as const,
|
||||
content: [{ type: "text" as const, text }],
|
||||
timestamp: 100 + index,
|
||||
})),
|
||||
},
|
||||
});
|
||||
expect(outcome.ok).toBe(true);
|
||||
if (!outcome.ok) {
|
||||
throw new Error(`expected worker transcript commit, received ${outcome.reason}`);
|
||||
}
|
||||
const events = await Promise.all(eventPromises);
|
||||
const payloads = events.map((event) =>
|
||||
requireRecord(event.payload, "session.message payload"),
|
||||
);
|
||||
expect(payloads.map((payload) => payload.messageId)).toEqual(outcome.result.entryIds);
|
||||
expect(payloads.map((payload) => payload.messageSeq)).toEqual([1, 2, 3]);
|
||||
expect(
|
||||
payloads.map((payload) => {
|
||||
const message = requireRecord(payload.message, "session.message payload message");
|
||||
return requireRecord(message["__openclaw"], "session.message metadata").id;
|
||||
}),
|
||||
).toEqual(outcome.result.entryIds);
|
||||
expect(
|
||||
payloads.map((payload) => {
|
||||
const message = requireRecord(payload.message, "session.message payload message");
|
||||
return requireRecord(message["__openclaw"], "session.message metadata").seq;
|
||||
}),
|
||||
).toEqual([1, 2, 3]);
|
||||
} finally {
|
||||
ws.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("routes transcript-only SQLite marker updates to the matching session owner", async () => {
|
||||
const storePath = await createSessionStoreFile();
|
||||
await writeSessionStore({
|
||||
|
||||
@@ -14,6 +14,7 @@ import {
|
||||
type OpenClawStateDatabase,
|
||||
} from "../../state/openclaw-state-db.js";
|
||||
import type { WorkerInstallationArtifact } from "./bundle.js";
|
||||
import type { WorkerConnectionIdentity } from "./connection-identity.js";
|
||||
import { hashWorkerCredential } from "./credential.js";
|
||||
import {
|
||||
createWorkerEnvironmentService,
|
||||
@@ -108,6 +109,7 @@ describe("worker environment service", () => {
|
||||
serviceOptions: Partial<
|
||||
Pick<
|
||||
WorkerEnvironmentServiceOptions,
|
||||
| "applyTranscriptCommit"
|
||||
| "bootstrapCallTimeoutMs"
|
||||
| "providerCallTimeoutMs"
|
||||
| "resolveSshIdentity"
|
||||
@@ -216,6 +218,33 @@ describe("worker environment service", () => {
|
||||
};
|
||||
}
|
||||
|
||||
function seedAttachedIdentity(
|
||||
environmentId: string,
|
||||
sessionId: string,
|
||||
): WorkerConnectionIdentity {
|
||||
const ready = seedReady(environmentId);
|
||||
const attached = store.transition({
|
||||
environmentId,
|
||||
from: ready.state,
|
||||
to: "attached",
|
||||
patch: attachedPatch(environmentId, sessionId),
|
||||
});
|
||||
const credential = store.getCredential(environmentId);
|
||||
if (!credential || !attached.bootstrapReceipt) {
|
||||
throw new Error("attached worker fixture is incomplete");
|
||||
}
|
||||
return {
|
||||
environmentId,
|
||||
credentialHash: credential.credentialHash,
|
||||
bundleHash: credential.bundleHash,
|
||||
sessionId,
|
||||
ownerEpoch: attached.ownerEpoch,
|
||||
rpcSetVersion: credential.rpcSetVersion,
|
||||
protocolFeatures: [...attached.bootstrapReceipt.protocolFeatures],
|
||||
credentialExpiresAtMs: credential.expiresAtMs,
|
||||
};
|
||||
}
|
||||
|
||||
it("persists intent and an immutable profile snapshot before provisioning", async () => {
|
||||
const operationIds: string[] = [];
|
||||
const provider = createProvider({
|
||||
@@ -318,6 +347,50 @@ describe("worker environment service", () => {
|
||||
expect(prepareInstallation).toHaveBeenCalledWith("bundle");
|
||||
});
|
||||
|
||||
it("fences transcript commits by current epoch and exact session credential binding", async () => {
|
||||
const environmentId = "worker-transcript-fence";
|
||||
const sessionId = "session-transcript-fence";
|
||||
const identity = seedAttachedIdentity(environmentId, sessionId);
|
||||
const applyTranscriptCommit = vi.fn(async () => ({
|
||||
ok: true as const,
|
||||
result: { entryIds: ["entry-1"], newLeafId: "entry-1" },
|
||||
}));
|
||||
const workerService = createService(createProvider(), { applyTranscriptCommit });
|
||||
const request = {
|
||||
runEpoch: identity.ownerEpoch,
|
||||
seq: 1,
|
||||
baseLeafId: null,
|
||||
messages: [
|
||||
{
|
||||
role: "user" as const,
|
||||
content: [{ type: "text" as const, text: "hello" }],
|
||||
timestamp: 1,
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
await expect(workerService.commitTranscript(identity, request)).resolves.toMatchObject({
|
||||
ok: true,
|
||||
});
|
||||
expect(applyTranscriptCommit).toHaveBeenCalledOnce();
|
||||
|
||||
await expect(
|
||||
workerService.commitTranscript(identity, {
|
||||
...request,
|
||||
runEpoch: identity.ownerEpoch + 1,
|
||||
seq: 2,
|
||||
}),
|
||||
).resolves.toEqual({ ok: false, reason: "epoch-mismatch" });
|
||||
|
||||
database.db
|
||||
.prepare("UPDATE worker_environment_credentials SET session_id = ? WHERE environment_id = ?")
|
||||
.run("session-other", environmentId);
|
||||
await expect(workerService.commitTranscript(identity, { ...request, seq: 2 })).resolves.toEqual(
|
||||
{ ok: false, reason: "session-not-attached" },
|
||||
);
|
||||
expect(applyTranscriptCommit).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("rejects attach before current bootstrap", async () => {
|
||||
const staleId = "worker-stale-attach";
|
||||
const bootstrapping = seedBootstrapping(staleId);
|
||||
|
||||
@@ -3,6 +3,10 @@ import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
|
||||
import {
|
||||
type WorkerAdmissionHandshake,
|
||||
type WorkerConnectParams,
|
||||
type WorkerProtocolCloseReason,
|
||||
type WorkerTranscriptCommitErrorReason,
|
||||
type WorkerTranscriptCommitParams,
|
||||
type WorkerTranscriptCommitResult,
|
||||
WORKER_RPC_SET_VERSION,
|
||||
} from "../../../packages/gateway-protocol/src/schema/worker-admission.js";
|
||||
import type { OpenClawConfig } from "../../config/types.js";
|
||||
@@ -22,6 +26,7 @@ import {
|
||||
type WorkerSshEndpoint,
|
||||
type WorkerSshIdentity,
|
||||
} from "../../plugins/types.js";
|
||||
import { safeEqualSecret } from "../../security/secret-equal.js";
|
||||
import { runTasksWithConcurrency } from "../../utils/run-with-concurrency.js";
|
||||
import {
|
||||
admitWorkerConnection,
|
||||
@@ -98,8 +103,20 @@ export type WorkerEnvironmentServiceOptions = {
|
||||
resolveWorkerGateway?: () => { host: "127.0.0.1" | "::1"; port: number } | undefined;
|
||||
now?: () => number;
|
||||
logger?: { warn: (message: string) => void };
|
||||
applyTranscriptCommit?: (params: {
|
||||
identity: WorkerConnectionIdentity;
|
||||
request: WorkerTranscriptCommitParams;
|
||||
}) => Promise<WorkerTranscriptCommitApplicationResult>;
|
||||
};
|
||||
|
||||
export type WorkerTranscriptCommitApplicationResult =
|
||||
| { ok: true; result: WorkerTranscriptCommitResult }
|
||||
| { ok: false; reason: WorkerTranscriptCommitErrorReason };
|
||||
|
||||
export type WorkerTranscriptCommitServiceResult =
|
||||
| WorkerTranscriptCommitApplicationResult
|
||||
| { ok: false; closeReason: WorkerProtocolCloseReason };
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
@@ -946,6 +963,47 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
|
||||
return { checkedAtMs, credentialHash, grant };
|
||||
};
|
||||
|
||||
const commitTranscript = (
|
||||
identity: WorkerConnectionIdentity,
|
||||
request: WorkerTranscriptCommitParams,
|
||||
): Promise<WorkerTranscriptCommitServiceResult> =>
|
||||
withLock(identity.environmentId, async () => {
|
||||
if (stopping) {
|
||||
return { ok: false, closeReason: "environment-unavailable" };
|
||||
}
|
||||
const credential = store.getCredential(identity.environmentId);
|
||||
if (!credential || !safeEqualSecret(credential.credentialHash, identity.credentialHash)) {
|
||||
return { ok: false, closeReason: "credential-replaced" };
|
||||
}
|
||||
if (now() >= credential.expiresAtMs) {
|
||||
return { ok: false, closeReason: "credential-expired" };
|
||||
}
|
||||
const environment = store.get(identity.environmentId);
|
||||
if (!environment || environment.destroyRequestedAtMs !== null) {
|
||||
return { ok: false, closeReason: "environment-unavailable" };
|
||||
}
|
||||
if (
|
||||
request.runEpoch !== identity.ownerEpoch ||
|
||||
request.runEpoch !== credential.ownerEpoch ||
|
||||
request.runEpoch !== environment.ownerEpoch
|
||||
) {
|
||||
return { ok: false, reason: "epoch-mismatch" };
|
||||
}
|
||||
if (
|
||||
environment.state !== "attached" ||
|
||||
!identity.sessionId ||
|
||||
credential.sessionId !== identity.sessionId ||
|
||||
environment.attachedSessionIds.length !== 1 ||
|
||||
environment.attachedSessionIds[0] !== identity.sessionId
|
||||
) {
|
||||
return { ok: false, reason: "session-not-attached" };
|
||||
}
|
||||
if (!options.applyTranscriptCommit) {
|
||||
return { ok: false, closeReason: "gateway-unavailable" };
|
||||
}
|
||||
return await options.applyTranscriptCommit({ identity, request });
|
||||
});
|
||||
|
||||
return {
|
||||
list: () => store.list().map(project),
|
||||
get: (environmentId: string) => {
|
||||
@@ -983,6 +1041,7 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
|
||||
stopping
|
||||
? ("environment-unavailable" as const)
|
||||
: validateWorkerConnectionIdentity({ store, identity, nowMs: now() }),
|
||||
commitTranscript,
|
||||
attachSession,
|
||||
takeMintedCredential: (binding: WorkerCredentialBinding) =>
|
||||
readPendingCredential(binding)?.grant,
|
||||
|
||||
@@ -281,6 +281,51 @@ describe("worker environment store", () => {
|
||||
).toThrow("owner epoch changed");
|
||||
});
|
||||
|
||||
it("allocates globally distinct owner epochs when a session moves environments", () => {
|
||||
const makeReady = (environmentId: string, leaseId: string) => {
|
||||
const bootstrapping = seedBootstrapping(environmentId, leaseId);
|
||||
return store.transition({
|
||||
environmentId,
|
||||
from: bootstrapping.state,
|
||||
to: "ready",
|
||||
patch: readyPatch(),
|
||||
});
|
||||
};
|
||||
|
||||
const firstReady = makeReady("worker-owner-a", "lease-owner-a");
|
||||
const first = store.transition({
|
||||
environmentId: firstReady.environmentId,
|
||||
from: firstReady.state,
|
||||
to: "attached",
|
||||
patch: attachedPatch("shared-session", firstReady.environmentId),
|
||||
});
|
||||
store.transition({
|
||||
environmentId: first.environmentId,
|
||||
from: first.state,
|
||||
to: "idle",
|
||||
});
|
||||
database.db
|
||||
.prepare(
|
||||
`INSERT INTO worker_transcript_commit_heads (
|
||||
session_id, run_epoch, environment_id, next_seq, updated_at_ms
|
||||
) VALUES (?, ?, ?, 1, ?)`,
|
||||
)
|
||||
.run("shared-session", first.ownerEpoch, first.environmentId, nowMs);
|
||||
database.db
|
||||
.prepare("DELETE FROM worker_environments WHERE environment_id = ?")
|
||||
.run(first.environmentId);
|
||||
const secondReady = makeReady("worker-owner-b", "lease-owner-b");
|
||||
const second = store.transition({
|
||||
environmentId: secondReady.environmentId,
|
||||
from: secondReady.state,
|
||||
to: "attached",
|
||||
patch: attachedPatch("shared-session", secondReady.environmentId),
|
||||
});
|
||||
|
||||
expect(first.ownerEpoch).toBe(2);
|
||||
expect(second.ownerEpoch).toBeGreaterThan(first.ownerEpoch);
|
||||
});
|
||||
|
||||
it("rejects illegal, stale, and lease-incomplete transitions", () => {
|
||||
createIntent();
|
||||
expect(() =>
|
||||
|
||||
@@ -63,7 +63,10 @@ export type WorkerEnvironmentTransitionPatch = {
|
||||
lastError?: string | null;
|
||||
credential?: CredentialInput;
|
||||
};
|
||||
type WorkerDb = Pick<StateDatabase, "worker_environment_credentials" | "worker_environments">;
|
||||
type WorkerDb = Pick<
|
||||
StateDatabase,
|
||||
"worker_environment_credentials" | "worker_environments" | "worker_transcript_commit_heads"
|
||||
>;
|
||||
type Row = Selectable<WorkerEnvironments>;
|
||||
type RowUpdate = Updateable<WorkerEnvironments>;
|
||||
type CredentialRow = Selectable<WorkerEnvironmentCredentials>;
|
||||
@@ -285,6 +288,25 @@ function nextOwnerEpoch(ownerEpoch: number): number {
|
||||
}
|
||||
return next;
|
||||
}
|
||||
function nextGlobalOwnerEpoch(db: DatabaseSync): number {
|
||||
// Transcript commit identity is (session, epoch, seq), so an ownership
|
||||
// generation may never be reused when a session moves between environments.
|
||||
const latestEnvironment = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
query(db)
|
||||
.selectFrom("worker_environments")
|
||||
.select(({ fn }) => fn.max<number>("owner_epoch").as("owner_epoch")),
|
||||
);
|
||||
const latestTranscriptCommit = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
query(db)
|
||||
.selectFrom("worker_transcript_commit_heads")
|
||||
.select(({ fn }) => fn.max<number>("run_epoch").as("run_epoch")),
|
||||
);
|
||||
return nextOwnerEpoch(
|
||||
Math.max(latestEnvironment?.owner_epoch ?? 0, latestTranscriptCommit?.run_epoch ?? 0),
|
||||
);
|
||||
}
|
||||
function fromRow(row: Row): WorkerEnvironmentRecord {
|
||||
const record = {
|
||||
environmentId: row.environment_id,
|
||||
@@ -684,11 +706,9 @@ export function createWorkerEnvironmentStore(
|
||||
to === "orphaned");
|
||||
const ownerEpoch = acceptsBootstrapReceipt
|
||||
? Math.max(1, current.ownerEpoch)
|
||||
: acceptsAttachedCredential
|
||||
? nextOwnerEpoch(current.ownerEpoch)
|
||||
: ownerEndingTransition
|
||||
? nextOwnerEpoch(current.ownerEpoch)
|
||||
: current.ownerEpoch;
|
||||
: acceptsAttachedCredential || ownerEndingTransition
|
||||
? nextGlobalOwnerEpoch(db)
|
||||
: current.ownerEpoch;
|
||||
const record = update(db, environmentId, from, {
|
||||
lease_id: leaseId,
|
||||
ssh_host: sshEndpoint?.host ?? null,
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
} from "../../state/openclaw-state-db.js";
|
||||
import {
|
||||
createWorkerTranscriptCommitStore,
|
||||
type WorkerTranscriptCommitInput,
|
||||
type WorkerTranscriptCommitOutcome,
|
||||
type WorkerTranscriptCommitStore,
|
||||
} from "./transcript-commit-store.js";
|
||||
|
||||
const SUCCESS_OUTCOME: WorkerTranscriptCommitOutcome = {
|
||||
ok: true,
|
||||
result: { entryIds: ["entry-a", "entry-b"], newLeafId: "entry-b" },
|
||||
};
|
||||
const ERROR_OUTCOME: WorkerTranscriptCommitOutcome = {
|
||||
ok: false,
|
||||
reason: "stale-base-leaf",
|
||||
};
|
||||
const BASE_INPUT: WorkerTranscriptCommitInput = {
|
||||
environmentId: "worker-a",
|
||||
sessionId: "session-a",
|
||||
runEpoch: 4,
|
||||
seq: 1,
|
||||
requestHash: "a".repeat(64),
|
||||
};
|
||||
|
||||
describe("worker transcript commit store", () => {
|
||||
let root: string;
|
||||
let nowMs: number;
|
||||
let store: WorkerTranscriptCommitStore;
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await fs.mkdtemp(path.join(await fs.realpath(os.tmpdir()), "openclaw-worker-commit-"));
|
||||
nowMs = 1_000;
|
||||
const database = openOpenClawStateDatabase({ env: { OPENCLAW_STATE_DIR: root } });
|
||||
store = createWorkerTranscriptCommitStore({ database, now: () => nowMs });
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
await fs.rm(root, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("recovers pending work and replays a terminal result across reopen", () => {
|
||||
expect(store.begin(BASE_INPUT)).toEqual({ kind: "claimed" });
|
||||
expect(store.begin(BASE_INPUT)).toEqual({ kind: "recover" });
|
||||
|
||||
nowMs = 1_010;
|
||||
expect(store.complete({ ...BASE_INPUT, outcome: SUCCESS_OUTCOME })).toEqual(SUCCESS_OUTCOME);
|
||||
expect(store.begin(BASE_INPUT)).toEqual({ kind: "replay", outcome: SUCCESS_OUTCOME });
|
||||
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
const database = openOpenClawStateDatabase({ env: { OPENCLAW_STATE_DIR: root } });
|
||||
store = createWorkerTranscriptCommitStore({ database, now: () => nowMs });
|
||||
expect(store.begin(BASE_INPUT)).toEqual({ kind: "replay", outcome: SUCCESS_OUTCOME });
|
||||
});
|
||||
|
||||
it("rejects a tuple replayed with a different payload or environment", () => {
|
||||
expect(store.begin(BASE_INPUT)).toEqual({ kind: "claimed" });
|
||||
expect(store.begin({ ...BASE_INPUT, requestHash: "b".repeat(64) })).toEqual({
|
||||
kind: "rejected",
|
||||
reason: "conflict",
|
||||
});
|
||||
store.complete({ ...BASE_INPUT, outcome: SUCCESS_OUTCOME });
|
||||
expect(store.begin({ ...BASE_INPUT, requestHash: "b".repeat(64) })).toEqual({
|
||||
kind: "rejected",
|
||||
reason: "conflict",
|
||||
});
|
||||
expect(store.begin({ ...BASE_INPUT, environmentId: "worker-b" })).toEqual({
|
||||
kind: "rejected",
|
||||
reason: "conflict",
|
||||
});
|
||||
});
|
||||
|
||||
it("advances one ordered sequence only after terminal completion", () => {
|
||||
const second = { ...BASE_INPUT, seq: 2, requestHash: "b".repeat(64) };
|
||||
const third = { ...BASE_INPUT, seq: 3, requestHash: "c".repeat(64) };
|
||||
|
||||
expect(store.begin(second)).toEqual({
|
||||
kind: "rejected",
|
||||
reason: "out-of-order",
|
||||
expectedSeq: 1,
|
||||
});
|
||||
expect(store.begin(BASE_INPUT)).toEqual({ kind: "claimed" });
|
||||
expect(store.begin(second)).toEqual({
|
||||
kind: "rejected",
|
||||
reason: "out-of-order",
|
||||
expectedSeq: 1,
|
||||
});
|
||||
store.complete({ ...BASE_INPUT, outcome: SUCCESS_OUTCOME });
|
||||
expect(store.begin(third)).toEqual({
|
||||
kind: "rejected",
|
||||
reason: "out-of-order",
|
||||
expectedSeq: 2,
|
||||
});
|
||||
expect(store.begin(second)).toEqual({ kind: "claimed" });
|
||||
expect(store.complete({ ...second, outcome: ERROR_OUTCOME })).toEqual(ERROR_OUTCOME);
|
||||
expect(store.begin(second)).toEqual({ kind: "replay", outcome: ERROR_OUTCOME });
|
||||
expect(store.begin(third)).toEqual({ kind: "claimed" });
|
||||
});
|
||||
|
||||
it("keeps the first cached terminal outcome", () => {
|
||||
expect(() => store.complete({ ...BASE_INPUT, outcome: SUCCESS_OUTCOME })).toThrow(
|
||||
"must begin before terminal completion",
|
||||
);
|
||||
store.begin(BASE_INPUT);
|
||||
expect(store.complete({ ...BASE_INPUT, outcome: SUCCESS_OUTCOME })).toEqual(SUCCESS_OUTCOME);
|
||||
expect(store.complete({ ...BASE_INPUT, outcome: ERROR_OUTCOME })).toEqual(SUCCESS_OUTCOME);
|
||||
});
|
||||
|
||||
it("starts an independent sequence for a later owner epoch", () => {
|
||||
expect(store.begin(BASE_INPUT)).toEqual({ kind: "claimed" });
|
||||
store.complete({ ...BASE_INPUT, outcome: SUCCESS_OUTCOME });
|
||||
const replacement = {
|
||||
...BASE_INPUT,
|
||||
environmentId: "worker-b",
|
||||
runEpoch: BASE_INPUT.runEpoch + 1,
|
||||
};
|
||||
|
||||
expect(store.begin(replacement)).toEqual({ kind: "claimed" });
|
||||
expect(store.complete({ ...replacement, outcome: SUCCESS_OUTCOME })).toEqual(SUCCESS_OUTCOME);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,323 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import type { Insertable, Selectable } from "kysely";
|
||||
import type {
|
||||
WorkerTranscriptCommitErrorReason,
|
||||
WorkerTranscriptCommitResult,
|
||||
} from "../../../packages/gateway-protocol/src/schema/worker-admission.js";
|
||||
import {
|
||||
executeSqliteQuerySync,
|
||||
executeSqliteQueryTakeFirstSync,
|
||||
getNodeSqliteKysely,
|
||||
} from "../../infra/kysely-sync.js";
|
||||
import type {
|
||||
DB as StateDatabase,
|
||||
WorkerTranscriptCommitHeads,
|
||||
WorkerTranscriptCommits,
|
||||
} from "../../state/openclaw-state-db.generated.js";
|
||||
import {
|
||||
openOpenClawStateDatabase,
|
||||
runOpenClawStateWriteTransaction,
|
||||
type OpenClawStateDatabase,
|
||||
} from "../../state/openclaw-state-db.js";
|
||||
|
||||
type TranscriptCommitDb = Pick<
|
||||
StateDatabase,
|
||||
"worker_transcript_commit_heads" | "worker_transcript_commits"
|
||||
>;
|
||||
type HeadRow = Selectable<WorkerTranscriptCommitHeads>;
|
||||
type HeadInsert = Insertable<WorkerTranscriptCommitHeads>;
|
||||
type CommitRow = Selectable<WorkerTranscriptCommits>;
|
||||
type CommitInsert = Insertable<WorkerTranscriptCommits>;
|
||||
|
||||
export type WorkerTranscriptCommitInput = {
|
||||
environmentId: string;
|
||||
sessionId: string;
|
||||
runEpoch: number;
|
||||
seq: number;
|
||||
requestHash: string;
|
||||
};
|
||||
|
||||
export type WorkerTranscriptCommitOutcome =
|
||||
| { ok: true; result: WorkerTranscriptCommitResult }
|
||||
| { ok: false; reason: WorkerTranscriptCommitErrorReason };
|
||||
|
||||
export type WorkerTranscriptCommitBeginResult =
|
||||
| { kind: "claimed" }
|
||||
| { kind: "recover" }
|
||||
| { kind: "replay"; outcome: WorkerTranscriptCommitOutcome }
|
||||
| { kind: "rejected"; reason: "conflict" }
|
||||
| { kind: "rejected"; reason: "out-of-order"; expectedSeq: number };
|
||||
|
||||
type NormalizedCommitInput = WorkerTranscriptCommitInput & { nowMs: number };
|
||||
type ExistingCommitResult = Extract<
|
||||
WorkerTranscriptCommitBeginResult,
|
||||
{ kind: "recover" | "replay" | "rejected" }
|
||||
>;
|
||||
|
||||
const REQUEST_HASH_PATTERN = /^[a-f0-9]{64}$/u;
|
||||
|
||||
function required(value: unknown, field: string): string {
|
||||
if (typeof value !== "string" || !value.trim()) {
|
||||
throw new Error(`Worker transcript commit ${field} must be a non-empty string`);
|
||||
}
|
||||
return value.trim();
|
||||
}
|
||||
|
||||
function nonNegativeInteger(value: unknown, field: string): number {
|
||||
if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) {
|
||||
throw new Error(`Worker transcript commit ${field} must be a non-negative integer`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function positiveInteger(value: unknown, field: string): number {
|
||||
if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 1) {
|
||||
throw new Error(`Worker transcript commit ${field} must be a positive integer`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function normalizeRequestHash(value: unknown): string {
|
||||
if (typeof value !== "string" || !REQUEST_HASH_PATTERN.test(value)) {
|
||||
throw new Error("Worker transcript commit request hash must be lowercase SHA-256 hex");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function isNonEmptyStringArray(value: unknown): value is string[] {
|
||||
return (
|
||||
Array.isArray(value) &&
|
||||
value.length > 0 &&
|
||||
value.every((entry: unknown) => typeof entry === "string" && entry.length > 0)
|
||||
);
|
||||
}
|
||||
|
||||
function isCommitResult(value: unknown): value is WorkerTranscriptCommitResult {
|
||||
if (!isRecord(value) || !isNonEmptyStringArray(value.entryIds)) {
|
||||
return false;
|
||||
}
|
||||
return typeof value.newLeafId === "string" && value.newLeafId.length > 0;
|
||||
}
|
||||
|
||||
function isCommitErrorReason(value: unknown): value is WorkerTranscriptCommitErrorReason {
|
||||
return (
|
||||
value === "stale-base-leaf" ||
|
||||
value === "epoch-mismatch" ||
|
||||
value === "invalid-batch" ||
|
||||
value === "session-not-attached"
|
||||
);
|
||||
}
|
||||
|
||||
function parseOutcomeJson(value: string): WorkerTranscriptCommitOutcome {
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(value) as unknown;
|
||||
} catch (error) {
|
||||
throw new Error("Worker transcript commit cached outcome is invalid", { cause: error });
|
||||
}
|
||||
if (!isRecord(parsed)) {
|
||||
throw new Error("Worker transcript commit cached outcome is invalid");
|
||||
}
|
||||
if (parsed.ok === true && isCommitResult(parsed.result)) {
|
||||
return { ok: true, result: parsed.result };
|
||||
}
|
||||
if (parsed.ok === false && isCommitErrorReason(parsed.reason)) {
|
||||
return { ok: false, reason: parsed.reason };
|
||||
}
|
||||
throw new Error("Worker transcript commit cached outcome is invalid");
|
||||
}
|
||||
|
||||
function serializeOutcome(outcome: WorkerTranscriptCommitOutcome): string {
|
||||
const serialized = JSON.stringify(outcome);
|
||||
if (!serialized) {
|
||||
throw new Error("Worker transcript commit outcome is not serializable");
|
||||
}
|
||||
return serialized;
|
||||
}
|
||||
|
||||
function normalizeInput(input: WorkerTranscriptCommitInput, nowMs: number): NormalizedCommitInput {
|
||||
return {
|
||||
environmentId: required(input.environmentId, "environment id"),
|
||||
sessionId: required(input.sessionId, "session id"),
|
||||
runEpoch: nonNegativeInteger(input.runEpoch, "run epoch"),
|
||||
seq: positiveInteger(input.seq, "sequence"),
|
||||
requestHash: normalizeRequestHash(input.requestHash),
|
||||
nowMs: nonNegativeInteger(nowMs, "timestamp"),
|
||||
};
|
||||
}
|
||||
|
||||
const query = (db: DatabaseSync) => getNodeSqliteKysely<TranscriptCommitDb>(db);
|
||||
|
||||
function findHead(db: DatabaseSync, input: NormalizedCommitInput): HeadRow | undefined {
|
||||
return executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
query(db)
|
||||
.selectFrom("worker_transcript_commit_heads")
|
||||
.selectAll()
|
||||
.where("session_id", "=", input.sessionId)
|
||||
.where("run_epoch", "=", input.runEpoch),
|
||||
);
|
||||
}
|
||||
|
||||
function findCommit(db: DatabaseSync, input: NormalizedCommitInput): CommitRow | undefined {
|
||||
return executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
query(db)
|
||||
.selectFrom("worker_transcript_commits")
|
||||
.selectAll()
|
||||
.where("session_id", "=", input.sessionId)
|
||||
.where("run_epoch", "=", input.runEpoch)
|
||||
.where("seq", "=", input.seq),
|
||||
);
|
||||
}
|
||||
|
||||
function classifyExistingCommit(params: {
|
||||
head: HeadRow | undefined;
|
||||
commit: CommitRow | undefined;
|
||||
input: NormalizedCommitInput;
|
||||
}): ExistingCommitResult | undefined {
|
||||
if (!params.commit) {
|
||||
return undefined;
|
||||
}
|
||||
if (!params.head) {
|
||||
throw new Error("Worker transcript commit row has no sequence head");
|
||||
}
|
||||
if (
|
||||
params.head.environment_id !== params.input.environmentId ||
|
||||
params.commit.request_hash !== params.input.requestHash
|
||||
) {
|
||||
return { kind: "rejected", reason: "conflict" };
|
||||
}
|
||||
if (params.commit.state === "pending") {
|
||||
return { kind: "recover" };
|
||||
}
|
||||
if (params.commit.state === "terminal" && params.commit.result_json !== null) {
|
||||
return { kind: "replay", outcome: parseOutcomeJson(params.commit.result_json) };
|
||||
}
|
||||
throw new Error("Worker transcript commit row has invalid terminal state");
|
||||
}
|
||||
|
||||
function insertHead(db: DatabaseSync, input: NormalizedCommitInput): void {
|
||||
const head: HeadInsert = {
|
||||
session_id: input.sessionId,
|
||||
run_epoch: input.runEpoch,
|
||||
environment_id: input.environmentId,
|
||||
next_seq: 1,
|
||||
updated_at_ms: input.nowMs,
|
||||
};
|
||||
executeSqliteQuerySync(db, query(db).insertInto("worker_transcript_commit_heads").values(head));
|
||||
}
|
||||
|
||||
function insertPendingCommit(db: DatabaseSync, input: NormalizedCommitInput): void {
|
||||
const commit: CommitInsert = {
|
||||
session_id: input.sessionId,
|
||||
run_epoch: input.runEpoch,
|
||||
seq: input.seq,
|
||||
request_hash: input.requestHash,
|
||||
state: "pending",
|
||||
result_json: null,
|
||||
created_at_ms: input.nowMs,
|
||||
updated_at_ms: input.nowMs,
|
||||
};
|
||||
executeSqliteQuerySync(db, query(db).insertInto("worker_transcript_commits").values(commit));
|
||||
}
|
||||
|
||||
export function createWorkerTranscriptCommitStore(
|
||||
options: { database?: OpenClawStateDatabase; now?: () => number } = {},
|
||||
) {
|
||||
const path = (options.database ?? openOpenClawStateDatabase()).path;
|
||||
const now = options.now ?? Date.now;
|
||||
const write = <T>(operation: (db: DatabaseSync) => T): T =>
|
||||
runOpenClawStateWriteTransaction(({ db }) => operation(db), { path });
|
||||
|
||||
const begin = (rawInput: WorkerTranscriptCommitInput): WorkerTranscriptCommitBeginResult => {
|
||||
const input = normalizeInput(rawInput, now());
|
||||
return write<WorkerTranscriptCommitBeginResult>((db) => {
|
||||
const head = findHead(db, input);
|
||||
const existing = classifyExistingCommit({ head, commit: findCommit(db, input), input });
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
if (head && head.environment_id !== input.environmentId) {
|
||||
return { kind: "rejected", reason: "conflict" };
|
||||
}
|
||||
const expectedSeq = head?.next_seq ?? 1;
|
||||
if (input.seq !== expectedSeq) {
|
||||
return { kind: "rejected", reason: "out-of-order", expectedSeq };
|
||||
}
|
||||
if (!head) {
|
||||
insertHead(db, input);
|
||||
}
|
||||
insertPendingCommit(db, input);
|
||||
return { kind: "claimed" };
|
||||
});
|
||||
};
|
||||
|
||||
const complete = (
|
||||
rawInput: WorkerTranscriptCommitInput & { outcome: WorkerTranscriptCommitOutcome },
|
||||
): WorkerTranscriptCommitOutcome => {
|
||||
const input = normalizeInput(rawInput, now());
|
||||
const resultJson = serializeOutcome(rawInput.outcome);
|
||||
return write<WorkerTranscriptCommitOutcome>((db) => {
|
||||
const head = findHead(db, input);
|
||||
const commit = findCommit(db, input);
|
||||
const existing = classifyExistingCommit({ head, commit, input });
|
||||
if (!existing) {
|
||||
throw new Error("Worker transcript commit must begin before terminal completion");
|
||||
}
|
||||
if (existing.kind === "rejected") {
|
||||
throw new Error(
|
||||
`Worker transcript commit terminal completion rejected: ${existing.reason}`,
|
||||
);
|
||||
}
|
||||
if (existing.kind === "replay") {
|
||||
return existing.outcome;
|
||||
}
|
||||
if (!head) {
|
||||
throw new Error("Worker transcript commit row has no sequence head");
|
||||
}
|
||||
if (head.next_seq !== input.seq) {
|
||||
throw new Error(
|
||||
`Worker transcript commit terminal completion expected sequence ${head.next_seq}`,
|
||||
);
|
||||
}
|
||||
|
||||
const commitUpdate = executeSqliteQuerySync(
|
||||
db,
|
||||
query(db)
|
||||
.updateTable("worker_transcript_commits")
|
||||
.set({ state: "terminal", result_json: resultJson, updated_at_ms: input.nowMs })
|
||||
.where("session_id", "=", input.sessionId)
|
||||
.where("run_epoch", "=", input.runEpoch)
|
||||
.where("seq", "=", input.seq)
|
||||
.where("request_hash", "=", input.requestHash)
|
||||
.where("state", "=", "pending"),
|
||||
);
|
||||
if (commitUpdate.numAffectedRows !== 1n) {
|
||||
throw new Error("Worker transcript commit changed during terminal completion");
|
||||
}
|
||||
const headUpdate = executeSqliteQuerySync(
|
||||
db,
|
||||
query(db)
|
||||
.updateTable("worker_transcript_commit_heads")
|
||||
.set({ next_seq: input.seq + 1, updated_at_ms: input.nowMs })
|
||||
.where("session_id", "=", input.sessionId)
|
||||
.where("run_epoch", "=", input.runEpoch)
|
||||
.where("environment_id", "=", input.environmentId)
|
||||
.where("next_seq", "=", input.seq),
|
||||
);
|
||||
if (headUpdate.numAffectedRows !== 1n) {
|
||||
throw new Error("Worker transcript commit sequence changed during terminal completion");
|
||||
}
|
||||
return rawInput.outcome;
|
||||
});
|
||||
};
|
||||
|
||||
return { begin, complete };
|
||||
}
|
||||
|
||||
export type WorkerTranscriptCommitStore = ReturnType<typeof createWorkerTranscriptCommitStore>;
|
||||
@@ -0,0 +1,716 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type {
|
||||
WorkerTranscriptCommitParams,
|
||||
WorkerTranscriptMessage,
|
||||
} from "../../../packages/gateway-protocol/src/schema/worker-admission.js";
|
||||
import { SessionManager } from "../../agents/sessions/session-manager.js";
|
||||
import {
|
||||
loadSessionEntry,
|
||||
loadTranscriptEvents,
|
||||
resolveSessionTranscriptRuntimeTarget,
|
||||
upsertSessionEntry,
|
||||
} from "../../config/sessions/session-accessor.js";
|
||||
import type { OpenClawConfig } from "../../config/types.openclaw.js";
|
||||
import { onSessionTranscriptUpdate } from "../../sessions/transcript-events.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
} from "../../state/openclaw-state-db.js";
|
||||
import type { WorkerConnectionIdentity } from "./connection-identity.js";
|
||||
import {
|
||||
createWorkerTranscriptCommitStore,
|
||||
type WorkerTranscriptCommitStore,
|
||||
} from "./transcript-commit-store.js";
|
||||
import {
|
||||
createWorkerTranscriptCommitter,
|
||||
type WorkerTranscriptCommitter,
|
||||
} from "./transcript-commit.js";
|
||||
|
||||
const SESSION_ID = "session-worker-transcript";
|
||||
const SESSION_KEY = "agent:main:worker-transcript";
|
||||
const RUN_EPOCH = 7;
|
||||
|
||||
const IDENTITY: WorkerConnectionIdentity = {
|
||||
environmentId: "environment-a",
|
||||
credentialHash: ["credential", "hash", "a"].join("-"),
|
||||
bundleHash: "b".repeat(64),
|
||||
sessionId: SESSION_ID,
|
||||
ownerEpoch: RUN_EPOCH,
|
||||
rpcSetVersion: 1,
|
||||
protocolFeatures: ["worker-transcript-commit-v1"],
|
||||
credentialExpiresAtMs: 10_000,
|
||||
};
|
||||
|
||||
const ZERO_USAGE = {
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 0,
|
||||
cost: {
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
total: 0,
|
||||
},
|
||||
};
|
||||
|
||||
function createTurnMessages(userText = "Inspect the workspace"): WorkerTranscriptMessage[] {
|
||||
return [
|
||||
{
|
||||
role: "user",
|
||||
content: [{ type: "text", text: userText }],
|
||||
timestamp: 100,
|
||||
},
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "thinking", thinking: "I will inspect it." },
|
||||
{
|
||||
type: "toolCall",
|
||||
id: "call-read-1",
|
||||
name: "read",
|
||||
arguments: { path: "README.md" },
|
||||
},
|
||||
],
|
||||
api: "openai-responses",
|
||||
provider: "openai",
|
||||
model: "gpt-5.5",
|
||||
diagnostics: [
|
||||
{
|
||||
type: "provider-warning",
|
||||
timestamp: 201,
|
||||
error: { name: "", message: "diagnostic", stack: "", code: 0 },
|
||||
details: { empty: "", enabled: false },
|
||||
},
|
||||
],
|
||||
usage: ZERO_USAGE,
|
||||
stopReason: "toolUse",
|
||||
timestamp: 200,
|
||||
},
|
||||
{
|
||||
role: "toolResult",
|
||||
toolCallId: "call-read-1",
|
||||
toolName: "read",
|
||||
content: [{ type: "text", text: "Workspace ready." }],
|
||||
isError: false,
|
||||
timestamp: 300,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function createRequest(
|
||||
params: {
|
||||
baseLeafId?: string | null;
|
||||
messages?: WorkerTranscriptMessage[];
|
||||
seq?: number;
|
||||
} = {},
|
||||
): WorkerTranscriptCommitParams {
|
||||
return {
|
||||
runEpoch: RUN_EPOCH,
|
||||
seq: params.seq ?? 1,
|
||||
baseLeafId: params.baseLeafId ?? null,
|
||||
messages: params.messages ?? createTurnMessages(),
|
||||
};
|
||||
}
|
||||
|
||||
function messageIdempotencyKey(seq: number, index: number): string {
|
||||
const digest = createHash("sha256")
|
||||
.update([SESSION_ID, RUN_EPOCH, seq, index].join("\0"))
|
||||
.digest("base64url");
|
||||
return `worker-commit-${digest}`;
|
||||
}
|
||||
|
||||
function requireAppendableWorkerMessage(
|
||||
message: unknown,
|
||||
): Parameters<SessionManager["appendMessage"]>[0] {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
throw new Error("expected committed worker message");
|
||||
}
|
||||
const role = (message as { role?: unknown }).role;
|
||||
if (role !== "assistant" && role !== "toolResult" && role !== "user") {
|
||||
throw new Error("expected committed worker message");
|
||||
}
|
||||
return message as Parameters<SessionManager["appendMessage"]>[0];
|
||||
}
|
||||
|
||||
describe("worker transcript commit application", () => {
|
||||
let root: string;
|
||||
let sessionsDir: string;
|
||||
let storePath: string;
|
||||
let sessionFile: string;
|
||||
let cfg: OpenClawConfig;
|
||||
let committer: WorkerTranscriptCommitter;
|
||||
let ledgerStore: WorkerTranscriptCommitStore;
|
||||
let unsubscribe: (() => void) | undefined;
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await fs.mkdtemp(path.join(await fs.realpath(os.tmpdir()), "openclaw-worker-turn-"));
|
||||
sessionsDir = path.join(root, "agents", "main", "sessions");
|
||||
storePath = path.join(sessionsDir, "sessions.json");
|
||||
cfg = {
|
||||
agents: { list: [{ id: "main", default: true }] },
|
||||
session: {
|
||||
mainKey: "main",
|
||||
store: path.join(root, "agents", "{agentId}", "sessions", "sessions.json"),
|
||||
},
|
||||
};
|
||||
await upsertSessionEntry(
|
||||
{ agentId: "main", sessionKey: SESSION_KEY, storePath },
|
||||
{
|
||||
sessionId: SESSION_ID,
|
||||
updatedAt: 10,
|
||||
},
|
||||
);
|
||||
sessionFile = (
|
||||
await resolveSessionTranscriptRuntimeTarget({
|
||||
agentId: "main",
|
||||
sessionId: SESSION_ID,
|
||||
sessionKey: SESSION_KEY,
|
||||
storePath,
|
||||
})
|
||||
).sessionFile;
|
||||
const database = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: path.join(root, "state") },
|
||||
});
|
||||
ledgerStore = createWorkerTranscriptCommitStore({ database });
|
||||
committer = createWorkerTranscriptCommitter({
|
||||
getConfig: () => cfg,
|
||||
store: ledgerStore,
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
unsubscribe?.();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
await fs.rm(root, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("commits semantic turns as a generated parent-linked transcript and publishes normally", async () => {
|
||||
const updates: Parameters<Parameters<typeof onSessionTranscriptUpdate>[0]>[0][] = [];
|
||||
unsubscribe = onSessionTranscriptUpdate((update) => updates.push(update));
|
||||
|
||||
const outcome = await committer.commit({ identity: IDENTITY, request: createRequest() });
|
||||
|
||||
expect(outcome.ok).toBe(true);
|
||||
if (!outcome.ok) {
|
||||
throw new Error(`expected transcript commit success, received ${outcome.reason}`);
|
||||
}
|
||||
const { entryIds, newLeafId } = outcome.result;
|
||||
expect(entryIds).toHaveLength(3);
|
||||
expect(new Set(entryIds).size).toBe(3);
|
||||
expect(newLeafId).toBe(entryIds[2]);
|
||||
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getLeafId()).toBe(newLeafId);
|
||||
expect(reopened.getEntries()).toEqual([
|
||||
expect.objectContaining({
|
||||
type: "message",
|
||||
id: entryIds[0],
|
||||
parentId: null,
|
||||
message: expect.objectContaining({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Inspect the workspace" }],
|
||||
}),
|
||||
}),
|
||||
expect.objectContaining({
|
||||
type: "message",
|
||||
id: entryIds[1],
|
||||
parentId: entryIds[0],
|
||||
message: expect.objectContaining({
|
||||
role: "assistant",
|
||||
content: expect.arrayContaining([
|
||||
expect.objectContaining({ type: "toolCall", id: "call-read-1" }),
|
||||
]),
|
||||
diagnostics: [
|
||||
{
|
||||
type: "provider-warning",
|
||||
timestamp: 201,
|
||||
error: { name: "", message: "diagnostic", stack: "", code: 0 },
|
||||
details: { empty: "", enabled: false },
|
||||
},
|
||||
],
|
||||
}),
|
||||
}),
|
||||
expect.objectContaining({
|
||||
type: "message",
|
||||
id: entryIds[2],
|
||||
parentId: entryIds[1],
|
||||
message: expect.objectContaining({
|
||||
role: "toolResult",
|
||||
toolCallId: "call-read-1",
|
||||
}),
|
||||
}),
|
||||
]);
|
||||
|
||||
const readEvents = await loadTranscriptEvents({
|
||||
agentId: "main",
|
||||
sessionId: SESSION_ID,
|
||||
sessionKey: SESSION_KEY,
|
||||
storePath,
|
||||
});
|
||||
expect(
|
||||
readEvents
|
||||
.filter((event): event is { type: "message"; id: string } =>
|
||||
Boolean(
|
||||
event &&
|
||||
typeof event === "object" &&
|
||||
!Array.isArray(event) &&
|
||||
(event as { type?: unknown }).type === "message" &&
|
||||
typeof (event as { id?: unknown }).id === "string",
|
||||
),
|
||||
)
|
||||
.map((event) => event.id),
|
||||
).toEqual(entryIds);
|
||||
const persistedEntry = loadSessionEntry({
|
||||
agentId: "main",
|
||||
sessionKey: SESSION_KEY,
|
||||
storePath,
|
||||
});
|
||||
expect(persistedEntry).toMatchObject({ sessionId: SESSION_ID });
|
||||
expect(persistedEntry?.sessionFile).toBe(
|
||||
(
|
||||
await resolveSessionTranscriptRuntimeTarget({
|
||||
agentId: "main",
|
||||
sessionId: SESSION_ID,
|
||||
sessionKey: SESSION_KEY,
|
||||
storePath,
|
||||
})
|
||||
).sessionFile,
|
||||
);
|
||||
expect(updates).toEqual(
|
||||
entryIds.map((entryId, index) =>
|
||||
expect.objectContaining({
|
||||
agentId: "main",
|
||||
message: expect.objectContaining({ role: createTurnMessages()[index]?.role }),
|
||||
messageId: entryId,
|
||||
messageSeq: index + 1,
|
||||
sessionKey: SESSION_KEY,
|
||||
sessionId: SESSION_ID,
|
||||
target: {
|
||||
agentId: "main",
|
||||
sessionId: SESSION_ID,
|
||||
sessionKey: SESSION_KEY,
|
||||
},
|
||||
}),
|
||||
),
|
||||
);
|
||||
});
|
||||
|
||||
it("durably materializes a user-only commit", async () => {
|
||||
const outcome = await committer.commit({
|
||||
identity: IDENTITY,
|
||||
request: createRequest({
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Persist before inference" }],
|
||||
timestamp: 100,
|
||||
},
|
||||
],
|
||||
}),
|
||||
});
|
||||
|
||||
expect(outcome.ok).toBe(true);
|
||||
if (!outcome.ok) {
|
||||
throw new Error(`expected user-only transcript commit, received ${outcome.reason}`);
|
||||
}
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getEntries()).toEqual([
|
||||
expect.objectContaining({
|
||||
id: outcome.result.newLeafId,
|
||||
parentId: null,
|
||||
message: expect.objectContaining({ role: "user" }),
|
||||
}),
|
||||
]);
|
||||
expect(reopened.getLeafId()).toBe(outcome.result.newLeafId);
|
||||
});
|
||||
|
||||
it("rejects a stale base leaf without appending", async () => {
|
||||
const first = await committer.commit({ identity: IDENTITY, request: createRequest() });
|
||||
if (!first.ok) {
|
||||
throw new Error(`expected initial transcript commit success, received ${first.reason}`);
|
||||
}
|
||||
|
||||
const stale = await committer.commit({
|
||||
identity: IDENTITY,
|
||||
request: createRequest({
|
||||
baseLeafId: null,
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Stale turn" }],
|
||||
timestamp: 400,
|
||||
},
|
||||
],
|
||||
seq: 2,
|
||||
}),
|
||||
});
|
||||
|
||||
expect(stale).toEqual({ ok: false, reason: "stale-base-leaf" });
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getEntries()).toHaveLength(3);
|
||||
expect(reopened.getLeafId()).toBe(first.result.newLeafId);
|
||||
});
|
||||
|
||||
it("replays the same tuple without duplicates and rejects a changed payload", async () => {
|
||||
const request = createRequest();
|
||||
const first = await committer.commit({ identity: IDENTITY, request });
|
||||
const replay = await committer.commit({
|
||||
identity: IDENTITY,
|
||||
request: structuredClone(request),
|
||||
});
|
||||
const changed = await committer.commit({
|
||||
identity: IDENTITY,
|
||||
request: createRequest({ messages: createTurnMessages("Changed payload") }),
|
||||
});
|
||||
|
||||
expect(first.ok).toBe(true);
|
||||
expect(replay).toEqual(first);
|
||||
expect(changed).toEqual({ ok: false, reason: "invalid-batch" });
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getEntries()).toHaveLength(3);
|
||||
if (first.ok) {
|
||||
expect(reopened.getLeafId()).toBe(first.result.newLeafId);
|
||||
}
|
||||
});
|
||||
|
||||
it("recovers an interrupted terminal write after later transcript activity", async () => {
|
||||
let interruptCompletion = true;
|
||||
const interruptedStore: WorkerTranscriptCommitStore = {
|
||||
begin: ledgerStore.begin,
|
||||
complete: (input) => {
|
||||
if (interruptCompletion) {
|
||||
interruptCompletion = false;
|
||||
throw new Error("simulated commit-result interruption");
|
||||
}
|
||||
return ledgerStore.complete(input);
|
||||
},
|
||||
};
|
||||
const interruptedCommitter = createWorkerTranscriptCommitter({
|
||||
getConfig: () => cfg,
|
||||
store: interruptedStore,
|
||||
});
|
||||
const request = createRequest();
|
||||
|
||||
await expect(interruptedCommitter.commit({ identity: IDENTITY, request })).rejects.toThrow(
|
||||
"simulated commit-result interruption",
|
||||
);
|
||||
const afterInterruption = SessionManager.open(sessionFile);
|
||||
const committedEntryIds = afterInterruption.getEntries().map((entry) => entry.id);
|
||||
expect(committedEntryIds).toHaveLength(request.messages.length);
|
||||
const laterLeafId = afterInterruption.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Later local activity" }],
|
||||
timestamp: 400,
|
||||
});
|
||||
|
||||
const replay = await committer.commit({ identity: IDENTITY, request });
|
||||
|
||||
expect(replay).toEqual({
|
||||
ok: true,
|
||||
result: {
|
||||
entryIds: committedEntryIds,
|
||||
newLeafId: committedEntryIds.at(-1),
|
||||
},
|
||||
});
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getEntries()).toHaveLength(request.messages.length + 1);
|
||||
expect(reopened.getLeafId()).toBe(laterLeafId);
|
||||
});
|
||||
|
||||
it("replays an interrupted terminal write after its branch is abandoned", async () => {
|
||||
cfg = { ...cfg, logging: { redactSensitive: "tools" } };
|
||||
const initialManager = SessionManager.open(sessionFile);
|
||||
const baseLeafId = initialManager.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Local base" }],
|
||||
timestamp: 50,
|
||||
});
|
||||
let interruptCompletion = true;
|
||||
const interruptedStore: WorkerTranscriptCommitStore = {
|
||||
begin: ledgerStore.begin,
|
||||
complete: (input) => {
|
||||
if (interruptCompletion) {
|
||||
interruptCompletion = false;
|
||||
throw new Error("simulated off-branch terminal interruption");
|
||||
}
|
||||
return ledgerStore.complete(input);
|
||||
},
|
||||
};
|
||||
const interruptedCommitter = createWorkerTranscriptCommitter({
|
||||
getConfig: () => cfg,
|
||||
store: interruptedStore,
|
||||
});
|
||||
const request = createRequest({
|
||||
baseLeafId,
|
||||
messages: createTurnMessages("my key is sk-abcdef1234567890xyz"),
|
||||
});
|
||||
|
||||
await expect(interruptedCommitter.commit({ identity: IDENTITY, request })).rejects.toThrow(
|
||||
"simulated off-branch terminal interruption",
|
||||
);
|
||||
const afterInterruption = SessionManager.open(sessionFile);
|
||||
const committedEntries = afterInterruption
|
||||
.getEntries()
|
||||
.filter((entry) => entry.id !== baseLeafId);
|
||||
const committedEntryIds = committedEntries.map((entry) => entry.id);
|
||||
expect(committedEntryIds).toHaveLength(request.messages.length);
|
||||
expect(JSON.stringify(committedEntries)).not.toContain("sk-abcdef1234567890xyz");
|
||||
|
||||
const firstCommitted = committedEntries[0];
|
||||
if (firstCommitted?.type !== "message") {
|
||||
throw new Error("expected committed worker message");
|
||||
}
|
||||
afterInterruption.branch(baseLeafId);
|
||||
const duplicatePrefixId = afterInterruption.appendMessage(
|
||||
requireAppendableWorkerMessage(firstCommitted.message),
|
||||
{ idempotencyLookup: "caller-checked" },
|
||||
);
|
||||
afterInterruption.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Incomplete duplicate branch" }],
|
||||
timestamp: 350,
|
||||
});
|
||||
afterInterruption.branch(baseLeafId);
|
||||
const localLeafId = afterInterruption.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Local branch wins" }],
|
||||
timestamp: 400,
|
||||
});
|
||||
const updates: Parameters<Parameters<typeof onSessionTranscriptUpdate>[0]>[0][] = [];
|
||||
unsubscribe = onSessionTranscriptUpdate((update) => updates.push(update));
|
||||
cfg = { ...cfg, logging: { redactSensitive: "off" } };
|
||||
|
||||
const replay = await committer.commit({ identity: IDENTITY, request });
|
||||
|
||||
expect(replay).toEqual({
|
||||
ok: true,
|
||||
result: {
|
||||
entryIds: committedEntryIds,
|
||||
newLeafId: committedEntryIds.at(-1),
|
||||
},
|
||||
});
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getBranch().map((entry) => entry.id)).toEqual([baseLeafId, localLeafId]);
|
||||
if (!replay.ok) {
|
||||
throw new Error(`expected interrupted commit replay, received ${replay.reason}`);
|
||||
}
|
||||
expect(replay.result.entryIds).not.toContain(duplicatePrefixId);
|
||||
expect(updates).toEqual([]);
|
||||
});
|
||||
|
||||
it("rejects ambiguous persisted recovery without appending or publishing", async () => {
|
||||
const initialManager = SessionManager.open(sessionFile);
|
||||
const baseLeafId = initialManager.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Local base" }],
|
||||
timestamp: 50,
|
||||
});
|
||||
let interruptCompletion = true;
|
||||
const interruptedCommitter = createWorkerTranscriptCommitter({
|
||||
getConfig: () => cfg,
|
||||
store: {
|
||||
begin: ledgerStore.begin,
|
||||
complete: (input) => {
|
||||
if (interruptCompletion) {
|
||||
interruptCompletion = false;
|
||||
throw new Error("simulated ambiguous terminal interruption");
|
||||
}
|
||||
return ledgerStore.complete(input);
|
||||
},
|
||||
},
|
||||
});
|
||||
const request = createRequest({ baseLeafId });
|
||||
|
||||
await expect(interruptedCommitter.commit({ identity: IDENTITY, request })).rejects.toThrow(
|
||||
"simulated ambiguous terminal interruption",
|
||||
);
|
||||
const manager = SessionManager.open(sessionFile);
|
||||
const originalEntries = manager.getEntries().filter((entry) => entry.id !== baseLeafId);
|
||||
expect(originalEntries).toHaveLength(request.messages.length);
|
||||
manager.branch(baseLeafId);
|
||||
for (const entry of originalEntries) {
|
||||
if (entry.type !== "message") {
|
||||
throw new Error("expected committed worker message");
|
||||
}
|
||||
manager.appendMessage(requireAppendableWorkerMessage(entry.message), {
|
||||
idempotencyLookup: "caller-checked",
|
||||
});
|
||||
}
|
||||
manager.branch(baseLeafId);
|
||||
const localLeafId = manager.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Local branch wins" }],
|
||||
timestamp: 400,
|
||||
});
|
||||
const entryCountBeforeRetry = manager.getEntries().length;
|
||||
const updates: Parameters<Parameters<typeof onSessionTranscriptUpdate>[0]>[0][] = [];
|
||||
unsubscribe = onSessionTranscriptUpdate((update) => updates.push(update));
|
||||
|
||||
const replay = await committer.commit({ identity: IDENTITY, request });
|
||||
|
||||
expect(replay).toEqual({ ok: false, reason: "invalid-batch" });
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getEntries()).toHaveLength(entryCountBeforeRetry);
|
||||
expect(reopened.getBranch().map((entry) => entry.id)).toEqual([baseLeafId, localLeafId]);
|
||||
expect(updates).toEqual([]);
|
||||
});
|
||||
|
||||
it("rolls back every transcript row when a batch append is interrupted", async () => {
|
||||
type AppendMessage = (
|
||||
this: SessionManager,
|
||||
...args: Parameters<SessionManager["appendMessage"]>
|
||||
) => ReturnType<SessionManager["appendMessage"]>;
|
||||
const appendMessage = Object.getOwnPropertyDescriptor(SessionManager.prototype, "appendMessage")
|
||||
?.value as AppendMessage | undefined;
|
||||
if (!appendMessage) {
|
||||
throw new Error("SessionManager.appendMessage implementation is unavailable");
|
||||
}
|
||||
let appendCount = 0;
|
||||
const appendSpy = vi
|
||||
.spyOn(SessionManager.prototype, "appendMessage")
|
||||
.mockImplementation(function (this: SessionManager, message, options) {
|
||||
const messageId = appendMessage.call(this, message, options);
|
||||
appendCount += 1;
|
||||
if (appendCount === 2) {
|
||||
throw new Error("simulated mid-batch interruption");
|
||||
}
|
||||
return messageId;
|
||||
});
|
||||
const request = createRequest();
|
||||
const entryBeforeFailure = loadSessionEntry({
|
||||
agentId: "main",
|
||||
sessionKey: SESSION_KEY,
|
||||
storePath,
|
||||
});
|
||||
|
||||
try {
|
||||
await expect(committer.commit({ identity: IDENTITY, request })).rejects.toThrow(
|
||||
"simulated mid-batch interruption",
|
||||
);
|
||||
} finally {
|
||||
appendSpy.mockRestore();
|
||||
}
|
||||
expect(SessionManager.open(sessionFile).getEntries()).toEqual([]);
|
||||
const entryAfterFailure = loadSessionEntry({
|
||||
agentId: "main",
|
||||
sessionKey: SESSION_KEY,
|
||||
storePath,
|
||||
});
|
||||
expect(entryAfterFailure).toEqual(entryBeforeFailure);
|
||||
|
||||
const manager = SessionManager.open(sessionFile);
|
||||
const localLeafId = manager.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Local activity after interruption" }],
|
||||
timestamp: 400,
|
||||
});
|
||||
const retry = await committer.commit({ identity: IDENTITY, request });
|
||||
|
||||
expect(retry).toEqual({ ok: false, reason: "stale-base-leaf" });
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getEntries()).toEqual([
|
||||
expect.objectContaining({
|
||||
id: localLeafId,
|
||||
message: expect.objectContaining({ role: "user" }),
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not reuse an idempotency key from an abandoned transcript branch", async () => {
|
||||
const first = await committer.commit({ identity: IDENTITY, request: createRequest() });
|
||||
if (!first.ok) {
|
||||
throw new Error(`expected initial transcript commit success, received ${first.reason}`);
|
||||
}
|
||||
const manager = SessionManager.open(sessionFile);
|
||||
const abandonedMessage: Parameters<SessionManager["appendMessage"]>[0] & {
|
||||
idempotencyKey: string;
|
||||
} = {
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Abandoned worker-shaped row" }],
|
||||
timestamp: 400,
|
||||
idempotencyKey: messageIdempotencyKey(2, 0),
|
||||
};
|
||||
const abandonedId = manager.appendMessage(abandonedMessage);
|
||||
manager.branch(first.result.newLeafId);
|
||||
const activeLeafId = manager.appendMessage({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Active local row" }],
|
||||
timestamp: 500,
|
||||
});
|
||||
|
||||
const outcome = await committer.commit({
|
||||
identity: IDENTITY,
|
||||
request: createRequest({
|
||||
baseLeafId: activeLeafId,
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
content: [{ type: "text", text: "Fresh worker row" }],
|
||||
timestamp: 600,
|
||||
},
|
||||
],
|
||||
seq: 2,
|
||||
}),
|
||||
});
|
||||
|
||||
expect(outcome.ok).toBe(true);
|
||||
if (!outcome.ok) {
|
||||
throw new Error(`expected branch-safe transcript commit, received ${outcome.reason}`);
|
||||
}
|
||||
expect(outcome.result.newLeafId).not.toBe(abandonedId);
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getLeafId()).toBe(outcome.result.newLeafId);
|
||||
expect(reopened.getEntry(outcome.result.newLeafId)).toMatchObject({
|
||||
parentId: activeLeafId,
|
||||
message: expect.objectContaining({ idempotencyKey: messageIdempotencyKey(2, 0) }),
|
||||
});
|
||||
});
|
||||
|
||||
it("advances the leaf across sequential commits", async () => {
|
||||
const first = await committer.commit({ identity: IDENTITY, request: createRequest() });
|
||||
if (!first.ok) {
|
||||
throw new Error(`expected initial transcript commit success, received ${first.reason}`);
|
||||
}
|
||||
const nextMessage: WorkerTranscriptMessage = {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "Finished." }],
|
||||
api: "openai-responses",
|
||||
provider: "openai",
|
||||
model: "gpt-5.5",
|
||||
usage: ZERO_USAGE,
|
||||
stopReason: "stop",
|
||||
timestamp: 400,
|
||||
};
|
||||
|
||||
const second = await committer.commit({
|
||||
identity: IDENTITY,
|
||||
request: createRequest({
|
||||
baseLeafId: first.result.newLeafId,
|
||||
messages: [nextMessage],
|
||||
seq: 2,
|
||||
}),
|
||||
});
|
||||
|
||||
expect(second.ok).toBe(true);
|
||||
if (!second.ok) {
|
||||
throw new Error(`expected sequential transcript commit success, received ${second.reason}`);
|
||||
}
|
||||
expect(second.result.entryIds).toHaveLength(1);
|
||||
expect(second.result.newLeafId).toBe(second.result.entryIds[0]);
|
||||
expect(second.result.newLeafId).not.toBe(first.result.newLeafId);
|
||||
const reopened = SessionManager.open(sessionFile);
|
||||
expect(reopened.getEntries().at(-1)).toMatchObject({
|
||||
id: second.result.newLeafId,
|
||||
parentId: first.result.newLeafId,
|
||||
message: expect.objectContaining({ role: "assistant" }),
|
||||
});
|
||||
expect(reopened.getLeafId()).toBe(second.result.newLeafId);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,529 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import type {
|
||||
WorkerTranscriptCommitParams,
|
||||
WorkerTranscriptMessage,
|
||||
} from "../../../packages/gateway-protocol/src/schema/worker-admission.js";
|
||||
import type { AgentMessage } from "../../agents/runtime/index.js";
|
||||
import { SessionManager } from "../../agents/sessions/session-manager.js";
|
||||
import { stableStringify } from "../../agents/stable-stringify.js";
|
||||
import { redactTranscriptMessage } from "../../agents/transcript-redact.js";
|
||||
import {
|
||||
loadSessionEntry,
|
||||
publishTranscriptUpdate,
|
||||
replaceSessionEntrySync,
|
||||
type SessionTranscriptWriteScope,
|
||||
withTranscriptWriteTransaction,
|
||||
} from "../../config/sessions/session-accessor.js";
|
||||
import type { OpenClawConfig } from "../../config/types.openclaw.js";
|
||||
import { KeyedAsyncQueue } from "../../plugin-sdk/keyed-async-queue.js";
|
||||
import { resolveSessionIdMatchSelection } from "../../sessions/session-id-resolution.js";
|
||||
import {
|
||||
loadCombinedSessionStoreForGateway,
|
||||
resolveFreshestSessionEntryFromStoreKeys,
|
||||
resolveGatewaySessionStoreTargetWithStore,
|
||||
} from "../session-utils.js";
|
||||
import type { WorkerConnectionIdentity } from "./connection-identity.js";
|
||||
import {
|
||||
createWorkerTranscriptCommitStore,
|
||||
type WorkerTranscriptCommitInput,
|
||||
type WorkerTranscriptCommitOutcome,
|
||||
type WorkerTranscriptCommitStore,
|
||||
} from "./transcript-commit-store.js";
|
||||
|
||||
type WorkerTranscriptCommitterOptions = {
|
||||
getConfig: () => OpenClawConfig;
|
||||
store?: WorkerTranscriptCommitStore;
|
||||
};
|
||||
|
||||
type ResolvedWorkerTranscriptTarget = Omit<
|
||||
SessionTranscriptWriteScope,
|
||||
"sessionId" | "sessionKey"
|
||||
> & {
|
||||
sessionEntry: NonNullable<ReturnType<typeof resolveFreshestSessionEntryFromStoreKeys>>;
|
||||
sessionId: string;
|
||||
sessionKey: string;
|
||||
sessionStore: Record<
|
||||
string,
|
||||
NonNullable<ReturnType<typeof resolveFreshestSessionEntryFromStoreKeys>>
|
||||
>;
|
||||
};
|
||||
|
||||
type SemanticAgentMessage = Extract<AgentMessage, { role: "assistant" | "toolResult" | "user" }>;
|
||||
type CommittedAgentMessage = SemanticAgentMessage & { idempotencyKey: string };
|
||||
|
||||
type AppliedTranscriptMessage = {
|
||||
appended: boolean;
|
||||
message: CommittedAgentMessage;
|
||||
messageId: string;
|
||||
messageSeq?: number;
|
||||
};
|
||||
|
||||
type ApplyTranscriptCommitResult =
|
||||
| { ok: true; messages: AppliedTranscriptMessage[] }
|
||||
| { ok: false; reason: "invalid-batch" | "session-not-attached" | "stale-base-leaf" };
|
||||
|
||||
type PersistedCommitResolution =
|
||||
| { kind: "ambiguous" | "missing" }
|
||||
| { kind: "found"; messages: AppliedTranscriptMessage[] };
|
||||
|
||||
function cloneContentPart(
|
||||
part: WorkerTranscriptMessage["content"][number],
|
||||
): WorkerTranscriptMessage["content"][number] {
|
||||
if (part.type === "text") {
|
||||
return {
|
||||
type: "text",
|
||||
text: part.text,
|
||||
...(part.textSignature ? { textSignature: part.textSignature } : {}),
|
||||
};
|
||||
}
|
||||
if (part.type === "image") {
|
||||
return { type: "image", data: part.data, mimeType: part.mimeType };
|
||||
}
|
||||
if (part.type === "thinking") {
|
||||
return {
|
||||
type: "thinking",
|
||||
thinking: part.thinking,
|
||||
...(part.thinkingSignature ? { thinkingSignature: part.thinkingSignature } : {}),
|
||||
...(part.redacted === undefined ? {} : { redacted: part.redacted }),
|
||||
};
|
||||
}
|
||||
return {
|
||||
type: "toolCall",
|
||||
id: part.id,
|
||||
name: part.name,
|
||||
arguments: structuredClone(part.arguments),
|
||||
...(part.thoughtSignature ? { thoughtSignature: part.thoughtSignature } : {}),
|
||||
...(part.executionMode ? { executionMode: part.executionMode } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function buildCommittedMessage(
|
||||
message: WorkerTranscriptMessage,
|
||||
idempotencyKey: string,
|
||||
): CommittedAgentMessage {
|
||||
const content = message.content.map((part) => cloneContentPart(part));
|
||||
if (message.role === "user") {
|
||||
return {
|
||||
role: "user",
|
||||
content,
|
||||
timestamp: message.timestamp,
|
||||
idempotencyKey,
|
||||
} as CommittedAgentMessage;
|
||||
}
|
||||
if (message.role === "toolResult") {
|
||||
return {
|
||||
role: "toolResult",
|
||||
toolCallId: message.toolCallId,
|
||||
toolName: message.toolName,
|
||||
content,
|
||||
...(message.details === undefined ? {} : { details: structuredClone(message.details) }),
|
||||
isError: message.isError,
|
||||
timestamp: message.timestamp,
|
||||
idempotencyKey,
|
||||
} as CommittedAgentMessage;
|
||||
}
|
||||
return {
|
||||
role: "assistant",
|
||||
content,
|
||||
api: message.api,
|
||||
provider: message.provider,
|
||||
model: message.model,
|
||||
...(message.responseModel ? { responseModel: message.responseModel } : {}),
|
||||
...(message.responseId ? { responseId: message.responseId } : {}),
|
||||
...(message.diagnostics
|
||||
? {
|
||||
diagnostics: message.diagnostics.map((diagnostic) => ({
|
||||
type: diagnostic.type,
|
||||
timestamp: diagnostic.timestamp,
|
||||
...(diagnostic.error
|
||||
? {
|
||||
error: {
|
||||
...(diagnostic.error.name === undefined ? {} : { name: diagnostic.error.name }),
|
||||
message: diagnostic.error.message,
|
||||
...(diagnostic.error.stack === undefined
|
||||
? {}
|
||||
: { stack: diagnostic.error.stack }),
|
||||
...(diagnostic.error.code === undefined ? {} : { code: diagnostic.error.code }),
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
...(diagnostic.details ? { details: structuredClone(diagnostic.details) } : {}),
|
||||
})),
|
||||
}
|
||||
: {}),
|
||||
usage: {
|
||||
input: message.usage.input,
|
||||
output: message.usage.output,
|
||||
cacheRead: message.usage.cacheRead,
|
||||
cacheWrite: message.usage.cacheWrite,
|
||||
...(message.usage.contextUsage
|
||||
? { contextUsage: structuredClone(message.usage.contextUsage) }
|
||||
: {}),
|
||||
totalTokens: message.usage.totalTokens,
|
||||
cost: {
|
||||
input: message.usage.cost.input,
|
||||
output: message.usage.cost.output,
|
||||
cacheRead: message.usage.cost.cacheRead,
|
||||
cacheWrite: message.usage.cost.cacheWrite,
|
||||
total: message.usage.cost.total,
|
||||
...(message.usage.cost.totalOrigin ? { totalOrigin: message.usage.cost.totalOrigin } : {}),
|
||||
},
|
||||
},
|
||||
stopReason: message.stopReason,
|
||||
...(message.errorMessage === undefined ? {} : { errorMessage: message.errorMessage }),
|
||||
...(message.errorCode === undefined ? {} : { errorCode: message.errorCode }),
|
||||
...(message.errorType === undefined ? {} : { errorType: message.errorType }),
|
||||
...(message.errorBody === undefined ? {} : { errorBody: message.errorBody }),
|
||||
timestamp: message.timestamp,
|
||||
idempotencyKey,
|
||||
} as CommittedAgentMessage;
|
||||
}
|
||||
|
||||
function requestHash(request: WorkerTranscriptCommitParams): string {
|
||||
return createHash("sha256")
|
||||
.update(
|
||||
stableStringify({
|
||||
baseLeafId: request.baseLeafId,
|
||||
messages: request.messages,
|
||||
}),
|
||||
)
|
||||
.digest("hex");
|
||||
}
|
||||
|
||||
function messageIdempotencyKey(params: {
|
||||
sessionId: string;
|
||||
runEpoch: number;
|
||||
seq: number;
|
||||
index: number;
|
||||
}): string {
|
||||
const digest = createHash("sha256")
|
||||
.update([params.sessionId, params.runEpoch, params.seq, params.index].join("\0"))
|
||||
.digest("base64url");
|
||||
return `worker-commit-${digest}`;
|
||||
}
|
||||
|
||||
function resolveWorkerTranscriptTarget(
|
||||
cfg: OpenClawConfig,
|
||||
sessionId: string,
|
||||
): ResolvedWorkerTranscriptTarget | undefined {
|
||||
const { store } = loadCombinedSessionStoreForGateway(cfg);
|
||||
const matches = Object.entries(store).filter(([, entry]) => entry.sessionId === sessionId);
|
||||
const selection = resolveSessionIdMatchSelection(matches, sessionId);
|
||||
if (selection.kind !== "selected") {
|
||||
return undefined;
|
||||
}
|
||||
const target = resolveGatewaySessionStoreTargetWithStore({
|
||||
cfg,
|
||||
key: selection.sessionKey,
|
||||
clone: false,
|
||||
});
|
||||
const entry = resolveFreshestSessionEntryFromStoreKeys(target.store, target.storeKeys);
|
||||
if (!entry || entry.sessionId !== sessionId) {
|
||||
return undefined;
|
||||
}
|
||||
return {
|
||||
agentId: target.agentId,
|
||||
sessionEntry: entry,
|
||||
sessionId,
|
||||
sessionKey: target.canonicalKey,
|
||||
sessionStore: target.store,
|
||||
storePath: target.storePath,
|
||||
};
|
||||
}
|
||||
|
||||
function readMessageIdempotencyKey(message: unknown): string | undefined {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
return undefined;
|
||||
}
|
||||
const value = (message as { idempotencyKey?: unknown }).idempotencyKey;
|
||||
return typeof value === "string" && value.trim() ? value.trim() : undefined;
|
||||
}
|
||||
|
||||
function isCommittedAgentMessage(message: unknown): message is CommittedAgentMessage {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
return false;
|
||||
}
|
||||
const role = (message as { role?: unknown }).role;
|
||||
return (
|
||||
(role === "user" || role === "assistant" || role === "toolResult") &&
|
||||
readMessageIdempotencyKey(message) !== undefined
|
||||
);
|
||||
}
|
||||
|
||||
function resolveActiveCommitPrefix(params: {
|
||||
baseLeafId: string | null;
|
||||
manager: SessionManager;
|
||||
messages: readonly CommittedAgentMessage[];
|
||||
}):
|
||||
| {
|
||||
activeVisibleEntryCount: number;
|
||||
ok: true;
|
||||
recoveredMessages: AppliedTranscriptMessage[];
|
||||
}
|
||||
| { ok: false } {
|
||||
const activeBranch = params.manager.getBranch();
|
||||
const activeVisibleEntryCount = activeBranch.filter(
|
||||
(entry) => entry.type === "message" || entry.type === "compaction",
|
||||
).length;
|
||||
if (params.manager.getLeafId() === params.baseLeafId) {
|
||||
return { activeVisibleEntryCount, ok: true, recoveredMessages: [] };
|
||||
}
|
||||
|
||||
const baseIndex =
|
||||
params.baseLeafId === null
|
||||
? -1
|
||||
: activeBranch.findIndex((entry) => entry.id === params.baseLeafId);
|
||||
if (params.baseLeafId !== null && baseIndex < 0) {
|
||||
return { ok: false };
|
||||
}
|
||||
|
||||
const activeSuffix = activeBranch.slice(baseIndex + 1);
|
||||
if (activeSuffix.length === 0) {
|
||||
return { ok: false };
|
||||
}
|
||||
|
||||
const recoveredMessages: AppliedTranscriptMessage[] = [];
|
||||
for (const [index, entry] of activeSuffix.slice(0, params.messages.length).entries()) {
|
||||
const expectedKey = readMessageIdempotencyKey(params.messages[index]);
|
||||
if (
|
||||
entry.type !== "message" ||
|
||||
!expectedKey ||
|
||||
!isCommittedAgentMessage(entry.message) ||
|
||||
readMessageIdempotencyKey(entry.message) !== expectedKey
|
||||
) {
|
||||
return { ok: false };
|
||||
}
|
||||
recoveredMessages.push({
|
||||
appended: false,
|
||||
message: entry.message,
|
||||
messageId: entry.id,
|
||||
});
|
||||
}
|
||||
return { activeVisibleEntryCount, ok: true, recoveredMessages };
|
||||
}
|
||||
|
||||
function resolvePersistedCommitAcrossDag(params: {
|
||||
baseLeafId: string | null;
|
||||
manager: SessionManager;
|
||||
messages: readonly CommittedAgentMessage[];
|
||||
}): PersistedCommitResolution {
|
||||
const childrenByParent = new Map<string | null, ReturnType<SessionManager["getEntries"]>>();
|
||||
for (const entry of params.manager.getEntries()) {
|
||||
const children = childrenByParent.get(entry.parentId) ?? [];
|
||||
children.push(entry);
|
||||
childrenByParent.set(entry.parentId, children);
|
||||
}
|
||||
|
||||
const completedPaths: AppliedTranscriptMessage[][] = [];
|
||||
const visit = (
|
||||
parentId: string | null,
|
||||
messageIndex: number,
|
||||
path: AppliedTranscriptMessage[],
|
||||
): void => {
|
||||
if (completedPaths.length > 1) {
|
||||
return;
|
||||
}
|
||||
if (messageIndex === params.messages.length) {
|
||||
completedPaths.push(path);
|
||||
return;
|
||||
}
|
||||
const expectedKey = readMessageIdempotencyKey(params.messages[messageIndex]);
|
||||
if (!expectedKey) {
|
||||
return;
|
||||
}
|
||||
for (const entry of childrenByParent.get(parentId) ?? []) {
|
||||
if (
|
||||
entry.type !== "message" ||
|
||||
!isCommittedAgentMessage(entry.message) ||
|
||||
readMessageIdempotencyKey(entry.message) !== expectedKey
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
visit(entry.id, messageIndex + 1, [
|
||||
...path,
|
||||
{ appended: false, message: entry.message, messageId: entry.id },
|
||||
]);
|
||||
}
|
||||
};
|
||||
|
||||
// The pending ledger binds the request hash while each deterministic key
|
||||
// binds tuple + index. Do not compare re-redacted content across restarts.
|
||||
visit(params.baseLeafId, 0, []);
|
||||
if (completedPaths.length > 1) {
|
||||
return { kind: "ambiguous" };
|
||||
}
|
||||
const messages = completedPaths[0];
|
||||
return messages ? { kind: "found", messages } : { kind: "missing" };
|
||||
}
|
||||
|
||||
async function applyWorkerTranscriptCommit(params: {
|
||||
config: OpenClawConfig;
|
||||
messages: readonly CommittedAgentMessage[];
|
||||
recoverPersistedBatch: boolean;
|
||||
requestedBaseLeafId: string | null;
|
||||
sessionId: string;
|
||||
target: ResolvedWorkerTranscriptTarget;
|
||||
}): Promise<ApplyTranscriptCommitResult> {
|
||||
const redactedMessages = params.messages.map(
|
||||
(message) => redactTranscriptMessage(message, params.config) as CommittedAgentMessage,
|
||||
);
|
||||
const applied = await withTranscriptWriteTransaction(params.target, ({ sessionFile }) => {
|
||||
const currentEntry = loadSessionEntry(params.target);
|
||||
if (!currentEntry || currentEntry.sessionId !== params.sessionId) {
|
||||
return { ok: false as const, reason: "session-not-attached" as const };
|
||||
}
|
||||
|
||||
const manager = SessionManager.open(sessionFile);
|
||||
if (params.recoverPersistedBatch) {
|
||||
// Only a pending ledger row may prove an off-branch batch: the agent DB
|
||||
// can commit before the shared replay ledger records its terminal result.
|
||||
const recovered = resolvePersistedCommitAcrossDag({
|
||||
baseLeafId: params.requestedBaseLeafId,
|
||||
manager,
|
||||
messages: redactedMessages,
|
||||
});
|
||||
if (recovered.kind === "found") {
|
||||
return { ok: true as const, messages: recovered.messages };
|
||||
}
|
||||
if (recovered.kind === "ambiguous") {
|
||||
return { ok: false as const, reason: "invalid-batch" as const };
|
||||
}
|
||||
}
|
||||
const prefix = resolveActiveCommitPrefix({
|
||||
baseLeafId: params.requestedBaseLeafId,
|
||||
manager,
|
||||
messages: redactedMessages,
|
||||
});
|
||||
if (!prefix.ok) {
|
||||
return { ok: false as const, reason: "stale-base-leaf" as const };
|
||||
}
|
||||
|
||||
const messages = [...prefix.recoveredMessages];
|
||||
let nextMessageSeq = prefix.activeVisibleEntryCount;
|
||||
for (const message of redactedMessages.slice(prefix.recoveredMessages.length)) {
|
||||
const messageId = manager.appendMessage(message, {
|
||||
config: params.config,
|
||||
// Active-path recovery owns dedupe. A global key scan could reuse an
|
||||
// id from an abandoned branch while SessionManager advances another id.
|
||||
idempotencyLookup: "caller-checked",
|
||||
});
|
||||
nextMessageSeq += 1;
|
||||
messages.push({
|
||||
appended: true,
|
||||
message,
|
||||
messageId,
|
||||
messageSeq: nextMessageSeq,
|
||||
});
|
||||
}
|
||||
|
||||
const freshEntry = loadSessionEntry(params.target);
|
||||
if (!freshEntry || freshEntry.sessionId !== params.sessionId) {
|
||||
return { ok: false as const, reason: "session-not-attached" as const };
|
||||
}
|
||||
const appendedCount = messages.filter((message) => message.appended).length;
|
||||
const nextEntry = {
|
||||
...freshEntry,
|
||||
sessionFile,
|
||||
...(appendedCount > 0 ? { updatedAt: Math.max(freshEntry.updatedAt ?? 0, Date.now()) } : {}),
|
||||
};
|
||||
replaceSessionEntrySync(params.target, nextEntry);
|
||||
return { ok: true as const, messages };
|
||||
});
|
||||
if (!applied.ok) {
|
||||
return applied;
|
||||
}
|
||||
|
||||
for (const message of applied.messages) {
|
||||
if (!message.appended) {
|
||||
continue;
|
||||
}
|
||||
await publishTranscriptUpdate(params.target, {
|
||||
message: message.message,
|
||||
messageId: message.messageId,
|
||||
messageSeq: message.messageSeq,
|
||||
});
|
||||
}
|
||||
return applied;
|
||||
}
|
||||
|
||||
/** Applies ordered, idempotent semantic worker turns to the canonical session transcript. */
|
||||
export function createWorkerTranscriptCommitter(options: WorkerTranscriptCommitterOptions) {
|
||||
const store = options.store ?? createWorkerTranscriptCommitStore();
|
||||
const sessionOperations = new KeyedAsyncQueue();
|
||||
|
||||
const commit = async (params: {
|
||||
identity: WorkerConnectionIdentity;
|
||||
request: WorkerTranscriptCommitParams;
|
||||
}): Promise<WorkerTranscriptCommitOutcome> => {
|
||||
const sessionId = params.identity.sessionId;
|
||||
if (!sessionId) {
|
||||
return { ok: false, reason: "session-not-attached" };
|
||||
}
|
||||
if (params.request.runEpoch !== params.identity.ownerEpoch) {
|
||||
return { ok: false, reason: "epoch-mismatch" };
|
||||
}
|
||||
return await sessionOperations.enqueue(sessionId, async () => {
|
||||
const input: WorkerTranscriptCommitInput = {
|
||||
environmentId: params.identity.environmentId,
|
||||
sessionId,
|
||||
runEpoch: params.request.runEpoch,
|
||||
seq: params.request.seq,
|
||||
requestHash: requestHash(params.request),
|
||||
};
|
||||
const started = store.begin(input);
|
||||
if (started.kind === "replay") {
|
||||
return started.outcome;
|
||||
}
|
||||
if (started.kind === "rejected") {
|
||||
return { ok: false, reason: "invalid-batch" };
|
||||
}
|
||||
|
||||
const config = options.getConfig();
|
||||
const target = resolveWorkerTranscriptTarget(config, sessionId);
|
||||
if (!target) {
|
||||
return store.complete({
|
||||
...input,
|
||||
outcome: { ok: false, reason: "session-not-attached" },
|
||||
});
|
||||
}
|
||||
const messages = params.request.messages.map((message, index) =>
|
||||
buildCommittedMessage(
|
||||
message,
|
||||
messageIdempotencyKey({
|
||||
sessionId,
|
||||
runEpoch: params.request.runEpoch,
|
||||
seq: params.request.seq,
|
||||
index,
|
||||
}),
|
||||
),
|
||||
);
|
||||
const applied = await applyWorkerTranscriptCommit({
|
||||
config,
|
||||
messages,
|
||||
recoverPersistedBatch: started.kind === "recover",
|
||||
requestedBaseLeafId: params.request.baseLeafId,
|
||||
sessionId,
|
||||
target,
|
||||
});
|
||||
if (!applied.ok) {
|
||||
return store.complete({ ...input, outcome: { ok: false, reason: applied.reason } });
|
||||
}
|
||||
const entryIds = applied.messages.map((message) => message.messageId);
|
||||
const newLeafId = entryIds.at(-1);
|
||||
if (entryIds.length !== params.request.messages.length || !newLeafId) {
|
||||
return store.complete({
|
||||
...input,
|
||||
outcome: { ok: false, reason: "invalid-batch" },
|
||||
});
|
||||
}
|
||||
return store.complete({
|
||||
...input,
|
||||
outcome: { ok: true, result: { entryIds, newLeafId } },
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
return { commit };
|
||||
}
|
||||
|
||||
export type WorkerTranscriptCommitter = ReturnType<typeof createWorkerTranscriptCommitter>;
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
OPENCLAW_AGENT_SCHEMA_VERSION,
|
||||
openOpenClawAgentDatabase,
|
||||
resolveOpenClawAgentSqlitePath,
|
||||
runOpenClawAgentWriteTransaction,
|
||||
} from "./openclaw-agent-db.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
@@ -966,6 +967,26 @@ describe("openclaw agent database", () => {
|
||||
expect(fs.statSync(parentDir).mode & 0o777).toBe(0o755);
|
||||
});
|
||||
|
||||
it.runIf(process.platform !== "win32")(
|
||||
"defers nested permission repair until the outer transaction commits",
|
||||
() => {
|
||||
const stateDir = createTempStateDir();
|
||||
const options = {
|
||||
agentId: "worker-1",
|
||||
env: { OPENCLAW_STATE_DIR: stateDir },
|
||||
};
|
||||
const database = openOpenClawAgentDatabase(options);
|
||||
fs.chmodSync(database.path, 0o644);
|
||||
|
||||
runOpenClawAgentWriteTransaction(() => {
|
||||
runOpenClawAgentWriteTransaction(() => undefined, options);
|
||||
expect(fs.statSync(database.path).mode & 0o777).toBe(0o644);
|
||||
}, options);
|
||||
|
||||
expect(fs.statSync(database.path).mode & 0o777).toBe(0o600);
|
||||
},
|
||||
);
|
||||
|
||||
it("configures durable SQLite connection pragmas", () => {
|
||||
const stateDir = createTempStateDir();
|
||||
const database = openOpenClawAgentDatabase({
|
||||
|
||||
@@ -831,13 +831,18 @@ export function runOpenClawAgentWriteTransaction<T>(
|
||||
> = {},
|
||||
): T {
|
||||
const database = openOpenClawAgentDatabase(options);
|
||||
const enteredNestedTransaction = database.db.isTransaction;
|
||||
const result = runSqliteImmediateTransactionSync(database.db, () => operation(database), {
|
||||
busyTimeoutMs: OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
|
||||
databaseLabel: database.path,
|
||||
...transactionOptions,
|
||||
operationLabel: transactionOptions.operationLabel ?? "agent.write",
|
||||
});
|
||||
ensureOpenClawAgentDatabasePermissions(database.path, options);
|
||||
// The outer owner repairs permissions after COMMIT; nested savepoint callers
|
||||
// must not add filesystem work while that transaction is still open.
|
||||
if (!enteredNestedTransaction) {
|
||||
ensureOpenClawAgentDatabasePermissions(database.path, options);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
+21
@@ -1135,6 +1135,25 @@ export interface WorkerEnvironments {
|
||||
updated_at_ms: number;
|
||||
}
|
||||
|
||||
export interface WorkerTranscriptCommitHeads {
|
||||
environment_id: string;
|
||||
next_seq: number;
|
||||
run_epoch: number;
|
||||
session_id: string;
|
||||
updated_at_ms: number;
|
||||
}
|
||||
|
||||
export interface WorkerTranscriptCommits {
|
||||
created_at_ms: number;
|
||||
request_hash: string;
|
||||
result_json: string | null;
|
||||
run_epoch: number;
|
||||
seq: number;
|
||||
session_id: string;
|
||||
state: string;
|
||||
updated_at_ms: number;
|
||||
}
|
||||
|
||||
export interface WorkspaceSetupState {
|
||||
bootstrap_seeded_at: string | null;
|
||||
setup_completed_at: string | null;
|
||||
@@ -1238,6 +1257,8 @@ export interface DB {
|
||||
web_push_vapid_keys: WebPushVapidKeys;
|
||||
worker_environment_credentials: WorkerEnvironmentCredentials;
|
||||
worker_environments: WorkerEnvironments;
|
||||
worker_transcript_commit_heads: WorkerTranscriptCommitHeads;
|
||||
worker_transcript_commits: WorkerTranscriptCommits;
|
||||
workspace_setup_state: WorkspaceSetupState;
|
||||
worktrees: Worktrees;
|
||||
}
|
||||
|
||||
@@ -843,6 +843,41 @@ describe("openclaw state database", () => {
|
||||
expect(credentialTable?.name).toBe("worker_environment_credentials");
|
||||
});
|
||||
|
||||
it("adds worker transcript commit tables to existing state databases", () => {
|
||||
const stateDir = createTempStateDir();
|
||||
const database = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: stateDir },
|
||||
});
|
||||
const databasePath = database.path;
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
|
||||
const { DatabaseSync } = requireNodeSqlite();
|
||||
const legacyDb = new DatabaseSync(databasePath);
|
||||
legacyDb.exec(`
|
||||
DROP TABLE worker_transcript_commits;
|
||||
DROP TABLE worker_transcript_commit_heads;
|
||||
`);
|
||||
legacyDb.close();
|
||||
|
||||
const reopened = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: stateDir },
|
||||
});
|
||||
const tables = reopened.db
|
||||
.prepare(
|
||||
`SELECT name FROM sqlite_master
|
||||
WHERE type = 'table' AND name IN (
|
||||
'worker_transcript_commit_heads',
|
||||
'worker_transcript_commits'
|
||||
)
|
||||
ORDER BY name`,
|
||||
)
|
||||
.all() as Array<{ name?: string }>;
|
||||
expect(tables.map((table) => table.name)).toEqual([
|
||||
"worker_transcript_commit_heads",
|
||||
"worker_transcript_commits",
|
||||
]);
|
||||
});
|
||||
|
||||
it("migrates requester and executor attribution for existing cross-agent tasks", () => {
|
||||
const stateDir = createTempStateDir();
|
||||
const database = openOpenClawStateDatabase({
|
||||
|
||||
@@ -1594,6 +1594,38 @@ CREATE TABLE IF NOT EXISTS worker_environment_credentials (
|
||||
FOREIGN KEY (environment_id) REFERENCES worker_environments(environment_id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
-- One durable sequence cursor per attached session owner epoch. The environment
|
||||
-- binding prevents independent workers with coincident epochs from sharing replay state.
|
||||
CREATE TABLE IF NOT EXISTS worker_transcript_commit_heads (
|
||||
session_id TEXT NOT NULL,
|
||||
run_epoch INTEGER NOT NULL CHECK (run_epoch >= 0),
|
||||
environment_id TEXT NOT NULL,
|
||||
next_seq INTEGER NOT NULL CHECK (next_seq >= 1),
|
||||
updated_at_ms INTEGER NOT NULL CHECK (updated_at_ms >= 0),
|
||||
PRIMARY KEY (session_id, run_epoch)
|
||||
);
|
||||
|
||||
-- Pending rows preserve a claimed request across gateway restarts. Terminal rows
|
||||
-- cache the exact result returned for deterministic at-least-once replay.
|
||||
CREATE TABLE IF NOT EXISTS worker_transcript_commits (
|
||||
session_id TEXT NOT NULL,
|
||||
run_epoch INTEGER NOT NULL CHECK (run_epoch >= 0),
|
||||
seq INTEGER NOT NULL CHECK (seq >= 1),
|
||||
request_hash TEXT NOT NULL,
|
||||
state TEXT NOT NULL CHECK (state IN ('pending', 'terminal')),
|
||||
result_json TEXT,
|
||||
created_at_ms INTEGER NOT NULL CHECK (created_at_ms >= 0),
|
||||
updated_at_ms INTEGER NOT NULL CHECK (updated_at_ms >= 0),
|
||||
PRIMARY KEY (session_id, run_epoch, seq),
|
||||
FOREIGN KEY (session_id, run_epoch)
|
||||
REFERENCES worker_transcript_commit_heads(session_id, run_epoch)
|
||||
ON DELETE CASCADE,
|
||||
CHECK (
|
||||
(state = 'pending' AND result_json IS NULL) OR
|
||||
(state = 'terminal' AND result_json IS NOT NULL)
|
||||
)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS fleet_cells (
|
||||
tenant_id TEXT NOT NULL PRIMARY KEY,
|
||||
created_at_ms INTEGER NOT NULL,
|
||||
|
||||
@@ -1589,6 +1589,38 @@ CREATE TABLE IF NOT EXISTS worker_environment_credentials (
|
||||
FOREIGN KEY (environment_id) REFERENCES worker_environments(environment_id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
-- One durable sequence cursor per attached session owner epoch. The environment
|
||||
-- binding prevents independent workers with coincident epochs from sharing replay state.
|
||||
CREATE TABLE IF NOT EXISTS worker_transcript_commit_heads (
|
||||
session_id TEXT NOT NULL,
|
||||
run_epoch INTEGER NOT NULL CHECK (run_epoch >= 0),
|
||||
environment_id TEXT NOT NULL,
|
||||
next_seq INTEGER NOT NULL CHECK (next_seq >= 1),
|
||||
updated_at_ms INTEGER NOT NULL CHECK (updated_at_ms >= 0),
|
||||
PRIMARY KEY (session_id, run_epoch)
|
||||
);
|
||||
|
||||
-- Pending rows preserve a claimed request across gateway restarts. Terminal rows
|
||||
-- cache the exact result returned for deterministic at-least-once replay.
|
||||
CREATE TABLE IF NOT EXISTS worker_transcript_commits (
|
||||
session_id TEXT NOT NULL,
|
||||
run_epoch INTEGER NOT NULL CHECK (run_epoch >= 0),
|
||||
seq INTEGER NOT NULL CHECK (seq >= 1),
|
||||
request_hash TEXT NOT NULL,
|
||||
state TEXT NOT NULL CHECK (state IN ('pending', 'terminal')),
|
||||
result_json TEXT,
|
||||
created_at_ms INTEGER NOT NULL CHECK (created_at_ms >= 0),
|
||||
updated_at_ms INTEGER NOT NULL CHECK (updated_at_ms >= 0),
|
||||
PRIMARY KEY (session_id, run_epoch, seq),
|
||||
FOREIGN KEY (session_id, run_epoch)
|
||||
REFERENCES worker_transcript_commit_heads(session_id, run_epoch)
|
||||
ON DELETE CASCADE,
|
||||
CHECK (
|
||||
(state = 'pending' AND result_json IS NULL) OR
|
||||
(state = 'terminal' AND result_json IS NOT NULL)
|
||||
)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS fleet_cells (
|
||||
tenant_id TEXT NOT NULL PRIMARY KEY,
|
||||
created_at_ms INTEGER NOT NULL,
|
||||
|
||||
Reference in New Issue
Block a user