mirror of
https://github.com/openclaw/openclaw.git
synced 2026-10-26 08:58:04 +00:00
fix: keep unrelated state handlers out of streamed reads (#182187)
* fix: keep unrelated state handlers out of streamed reads Prepare only the selected stream owner before fresh source admission. Keep mixed registry adapters out of emitted stream chunks and exercise both canonical SQLite flows through cold emitted workers. Scoped proof and P0 autoreview passed. The inherited shard root-budget failure and unreached import-closure validation remain recorded. No 1GiB/512MiB support or memory-reduction claim. * fix(state): preserve current restore semantics * fix(state): include stream helper in PR wrapper inventory
This commit is contained in:
Vincent Koc
GitHub
parent
c078b8d49a
commit
d57f91b8fa
11 changed files with 671 additions and 174 deletions
|
|
@ -2053,6 +2053,7 @@ src/state/openclaw-state-read-pool.ts
|
|||
src/state/openclaw-state-read-request.ts
|
||||
src/state/openclaw-state-read-scope.ts
|
||||
src/state/openclaw-state-read-snapshot-source.ts
|
||||
src/state/openclaw-state-read-stream.ts
|
||||
src/state/openclaw-state-read-worker.ts
|
||||
src/state/openclaw-state-read.worker.ts
|
||||
src/state/openclaw-state-schema-compatibility.ts
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ import {
|
|||
prepareSubagentRegistrationRows,
|
||||
prepareSubagentRunsSnapshotForRunIds,
|
||||
} from "./subagent-registry-state.js";
|
||||
import { streamSubagentRegistryInWorker } from "./subagent-registry.store.worker.js";
|
||||
import { streamSubagentRegistryInWorker } from "./subagent-registry.store.restore.worker.js";
|
||||
import type { SubagentRunRecord } from "./subagent-registry.types.js";
|
||||
import { isSameSubagentRunOwner } from "./subagent-run-generation.js";
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,74 @@
|
|||
import { getNodeSqliteKysely, iterateSqliteQuerySync } from "../../../infra/kysely-sync.js";
|
||||
import { isPostgresSqlConnection } from "../../../infra/sql-connection.js";
|
||||
import { isSqliteCorruptionError } from "../../../infra/sqlite-error-diagnostics.js";
|
||||
import { throwSqliteLifecycleErrors } from "../../../infra/sqlite-lifecycle-errors.js";
|
||||
import { admitSqliteSchema } from "../../../infra/sqlite-schema-facts.js";
|
||||
import type { WorkerTaskChannel } from "../../../infra/worker-task-server.js";
|
||||
import { invalidateOpenClawStateRuntimeIntegrity } from "../../../state/openclaw-state-db-integrity-admission.js";
|
||||
import {
|
||||
assertStateReadSchema,
|
||||
openOpenClawStateReadConnection,
|
||||
} from "../../../state/openclaw-state-db-read-connection.js";
|
||||
import type { DB as OpenClawStateKyselyDatabase } from "../../../state/openclaw-state-db.generated.js";
|
||||
import type { OpenClawStateReadRequest } from "../../../state/openclaw-state-read.types.js";
|
||||
import { rowToSubagentRunRecord } from "./subagent-registry.store.codec.js";
|
||||
|
||||
/** The single SELECT owns its snapshot; backpressure bounds decoded facts in flight. */
|
||||
export async function streamSubagentRegistryInWorker(
|
||||
input: OpenClawStateReadRequest,
|
||||
channel: WorkerTaskChannel,
|
||||
onAdmitted: () => void,
|
||||
): Promise<number> {
|
||||
const connection = openOpenClawStateReadConnection(
|
||||
input.databasePath,
|
||||
input.location,
|
||||
input.expectedIdentity,
|
||||
input.snapshotRoot,
|
||||
);
|
||||
const { db } = connection.database;
|
||||
const errors: unknown[] = [];
|
||||
let count = 0;
|
||||
try {
|
||||
assertStateReadSchema(db, input.databasePath);
|
||||
if (!isPostgresSqlConnection(db)) {
|
||||
admitSqliteSchema(db);
|
||||
}
|
||||
onAdmitted();
|
||||
const query = getNodeSqliteKysely<Pick<OpenClawStateKyselyDatabase, "subagent_runs">>(db)
|
||||
.selectFrom("subagent_runs")
|
||||
.selectAll()
|
||||
.orderBy("run_id", "asc");
|
||||
let batch = [];
|
||||
let bytes = 0;
|
||||
for (const row of iterateSqliteQuerySync(db, query)) {
|
||||
const size = Buffer.byteLength(row.payload_json);
|
||||
if (batch.length && (batch.length === 128 || bytes + size > 1024 * 1024)) {
|
||||
(await channel.request(batch)).consumed();
|
||||
batch = [];
|
||||
bytes = 0;
|
||||
}
|
||||
const entry = rowToSubagentRunRecord(row);
|
||||
if (!entry) {
|
||||
throw new Error("Canonical subagent restore found an unreadable durable row");
|
||||
}
|
||||
batch.push({ entry, createdAt: row.created_at });
|
||||
bytes += size;
|
||||
count += 1;
|
||||
}
|
||||
if (batch.length) {
|
||||
(await channel.request(batch)).consumed();
|
||||
}
|
||||
} catch (error) {
|
||||
if (!isPostgresSqlConnection(db) && isSqliteCorruptionError(error)) {
|
||||
invalidateOpenClawStateRuntimeIntegrity(db);
|
||||
}
|
||||
errors.push(error);
|
||||
}
|
||||
try {
|
||||
connection.close();
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
throwSqliteLifecycleErrors(errors, "Subagent restore and reader cleanup failed");
|
||||
return count;
|
||||
}
|
||||
|
|
@ -1,29 +1,14 @@
|
|||
import { requestSessionEntryCurrentAdmission } from "../../../config/sessions/session-entry-current-admission.worker.js";
|
||||
import { getNodeSqliteKysely, iterateSqliteQuerySync } from "../../../infra/kysely-sync.js";
|
||||
import { isPostgresSqlConnection, type EngineConnection } from "../../../infra/sql-connection.js";
|
||||
import { isSqliteCorruptionError } from "../../../infra/sqlite-error-diagnostics.js";
|
||||
import { throwSqliteLifecycleErrors } from "../../../infra/sqlite-lifecycle-errors.js";
|
||||
import type { EngineConnection } from "../../../infra/sql-connection.js";
|
||||
import { deferSqlitePostCommitPublication } from "../../../infra/sqlite-post-commit.js";
|
||||
import { admitSqliteSchema } from "../../../infra/sqlite-schema-facts.js";
|
||||
import {
|
||||
deferSqliteWorkerCommitReceipt,
|
||||
requestSqliteWorkerOperationAdmission,
|
||||
} from "../../../infra/sqlite-worker-operation-admission.js";
|
||||
import type { WorkerTaskChannel } from "../../../infra/worker-task-server.js";
|
||||
import { createSubsystemLogger } from "../../../logging/subsystem.js";
|
||||
import { recordSessionStateEventsInDatabase } from "../../../sessions/session-state-events.kernel.js";
|
||||
import { invalidateOpenClawStateRuntimeIntegrity } from "../../../state/openclaw-state-db-integrity-admission.js";
|
||||
import {
|
||||
assertStateReadSchema,
|
||||
openOpenClawStateReadConnection,
|
||||
} from "../../../state/openclaw-state-db-read-connection.js";
|
||||
import type { DB as OpenClawStateKyselyDatabase } from "../../../state/openclaw-state-db.generated.js";
|
||||
import type {
|
||||
OpenClawStateReadCommand,
|
||||
OpenClawStateReadRequest,
|
||||
} from "../../../state/openclaw-state-read.types.js";
|
||||
import type { OpenClawStateReadCommand } from "../../../state/openclaw-state-read.types.js";
|
||||
import type { WorkerWriteOperationContext } from "../../../state/worker-operation-registry.js";
|
||||
import { rowToSubagentRunRecord } from "./subagent-registry.store.codec.js";
|
||||
import { writeSubagentRunValuesInDatabase } from "./subagent-registry.store.kernel.js";
|
||||
import {
|
||||
loadSubagentMaintenanceRunsInDatabase,
|
||||
|
|
@ -88,66 +73,6 @@ export function readSubagentRunsInWorker(
|
|||
return { type: command.type, runs: new Map(rows.map((entry) => [entry.runId, entry])) };
|
||||
}
|
||||
|
||||
/** The single SELECT owns its snapshot; backpressure bounds decoded facts in flight. */
|
||||
export async function streamSubagentRegistryInWorker(
|
||||
input: OpenClawStateReadRequest,
|
||||
channel: WorkerTaskChannel,
|
||||
onAdmitted: () => void,
|
||||
): Promise<number> {
|
||||
const connection = openOpenClawStateReadConnection(
|
||||
input.databasePath,
|
||||
input.location,
|
||||
input.expectedIdentity,
|
||||
input.snapshotRoot,
|
||||
);
|
||||
const { db } = connection.database;
|
||||
const errors: unknown[] = [];
|
||||
let count = 0;
|
||||
try {
|
||||
assertStateReadSchema(db, input.databasePath);
|
||||
if (!isPostgresSqlConnection(db)) {
|
||||
admitSqliteSchema(db);
|
||||
}
|
||||
onAdmitted();
|
||||
const query = getNodeSqliteKysely<Pick<OpenClawStateKyselyDatabase, "subagent_runs">>(db)
|
||||
.selectFrom("subagent_runs")
|
||||
.selectAll()
|
||||
.orderBy("run_id", "asc");
|
||||
let batch = [];
|
||||
let bytes = 0;
|
||||
for (const row of iterateSqliteQuerySync(db, query)) {
|
||||
const size = Buffer.byteLength(row.payload_json);
|
||||
if (batch.length && (batch.length === 128 || bytes + size > 1024 * 1024)) {
|
||||
(await channel.request(batch)).consumed();
|
||||
batch = [];
|
||||
bytes = 0;
|
||||
}
|
||||
const entry = rowToSubagentRunRecord(row);
|
||||
if (!entry) {
|
||||
throw new Error("Canonical subagent restore found an unreadable durable row");
|
||||
}
|
||||
batch.push({ entry, createdAt: row.created_at });
|
||||
bytes += size;
|
||||
count += 1;
|
||||
}
|
||||
if (batch.length) {
|
||||
(await channel.request(batch)).consumed();
|
||||
}
|
||||
} catch (error) {
|
||||
if (!isPostgresSqlConnection(db) && isSqliteCorruptionError(error)) {
|
||||
invalidateOpenClawStateRuntimeIntegrity(db);
|
||||
}
|
||||
errors.push(error);
|
||||
}
|
||||
try {
|
||||
connection.close();
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
throwSqliteLifecycleErrors(errors, "Subagent restore and reader cleanup failed");
|
||||
return count;
|
||||
}
|
||||
|
||||
/** The row owner's worker adapter commits FIFO-planned rows and terminal signals together. */
|
||||
export const subagentRegistryOperations = {
|
||||
"subagents.persistChanges": (
|
||||
|
|
|
|||
|
|
@ -105,7 +105,6 @@ import { assertOpenClawStateWriteAllowed } from "./openclaw-state-ownership.js";
|
|||
import { readMcpOAuthStateCommand } from "./openclaw-state-read-mcp-oauth.js";
|
||||
import type { RegisteredStateReadCommand } from "./openclaw-state-read-operation-registry.js";
|
||||
import { readSessionRowsSharedFacts } from "./openclaw-state-read-session-rows.worker.js";
|
||||
import { readStateStream } from "./openclaw-state-read-stream.js";
|
||||
import type { OpenClawStateReadRequest } from "./openclaw-state-read.types.js";
|
||||
import { findSessionRepositoryWorkspaceInDatabase } from "./session-repository-workspaces.kernel.js";
|
||||
import { readUserBackgroundCommand } from "./user-background.kernel.js";
|
||||
|
|
@ -114,7 +113,7 @@ import { readUserPreferences, selectUserPreferenceValues } from "./user-preferen
|
|||
import { readUserProfileCommand } from "./user-profile-read.worker.js";
|
||||
import { selectResolvedUserProfileMetadataById } from "./user-profiles-internal.js";
|
||||
|
||||
export { prepareAcpSessionMetadataRead, loadSubagentSessionListRunsFromSqlite, readStateStream };
|
||||
export { prepareAcpSessionMetadataRead, loadSubagentSessionListRunsFromSqlite };
|
||||
|
||||
/** Non-metadata query owners load only when the shared worker receives one of their commands. */
|
||||
export function readOpenClawStateDatabaseCommand(
|
||||
|
|
|
|||
|
|
@ -1,13 +1,11 @@
|
|||
import { streamSubagentRegistryInWorker } from "../agents/subagents/registry/subagent-registry.store.worker.js";
|
||||
import type { WorkerTaskChannel } from "../infra/worker-task-server.js";
|
||||
import { streamTranscriptExportInWorker } from "../transcripts/store-export.worker.js";
|
||||
import type {
|
||||
OpenClawStateReadCommand,
|
||||
OpenClawStateReadReply,
|
||||
OpenClawStateReadRequest,
|
||||
} from "./openclaw-state-read.types.js";
|
||||
|
||||
export async function readStateStream(
|
||||
export async function prepareStateStream(
|
||||
input: OpenClawStateReadRequest,
|
||||
command: Extract<
|
||||
OpenClawStateReadCommand,
|
||||
|
|
@ -15,16 +13,25 @@ export async function readStateStream(
|
|||
>,
|
||||
channel: WorkerTaskChannel | undefined,
|
||||
onAdmitted: () => void,
|
||||
): Promise<OpenClawStateReadReply> {
|
||||
if (!channel) {
|
||||
const operation =
|
||||
command.type === "subagents.restore" ? "Subagent restore" : "Transcript export";
|
||||
throw new Error(`${operation} requires a bounded receiver`);
|
||||
}
|
||||
): Promise<() => Promise<OpenClawStateReadReply>> {
|
||||
if (command.type === "subagents.restore") {
|
||||
const count = await streamSubagentRegistryInWorker(input, channel, onAdmitted);
|
||||
return { ok: true, type: command.type, sourceAdmitted: true, count };
|
||||
// A separate restore entry keeps mixed read/write adapters out of the lazy worker graph.
|
||||
const { streamSubagentRegistryInWorker } =
|
||||
await import("../agents/subagents/registry/subagent-registry.store.restore.worker.js");
|
||||
return async () => {
|
||||
if (!channel) {
|
||||
throw new Error("Subagent restore requires a bounded receiver");
|
||||
}
|
||||
const count = await streamSubagentRegistryInWorker(input, channel, onAdmitted);
|
||||
return { ok: true, type: command.type, sourceAdmitted: true, count };
|
||||
};
|
||||
}
|
||||
const result = await streamTranscriptExportInWorker(input, command, channel, onAdmitted);
|
||||
return { ok: true, type: command.type, sourceAdmitted: true, result };
|
||||
const { streamTranscriptExportInWorker } = await import("../transcripts/store-export.worker.js");
|
||||
return async () => {
|
||||
if (!channel) {
|
||||
throw new Error("Transcript export requires a bounded receiver");
|
||||
}
|
||||
const result = await streamTranscriptExportInWorker(input, command, channel, onAdmitted);
|
||||
return { ok: true, type: command.type, sourceAdmitted: true, result };
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,11 @@
|
|||
import path from "node:path";
|
||||
import type { Worker } from "node:worker_threads";
|
||||
import { afterEach, beforeEach, expect, it, vi } from "vitest";
|
||||
import type {
|
||||
WorkerTaskChannel,
|
||||
WorkerTaskControl,
|
||||
WorkerTaskHandler,
|
||||
} from "@openclaw/worker-runtime/worker";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
|
||||
import type { PostgresPhysicalFacts } from "../infra/database-engine/physical-facts.js";
|
||||
import type { DatabaseEngineSelection } from "../infra/database-engine/types.js";
|
||||
|
|
@ -17,7 +22,19 @@ import type {
|
|||
const mock = vi.hoisted(() => {
|
||||
vi.resetModules();
|
||||
return {
|
||||
handler: vi.fn<(input: unknown) => Promise<OpenClawStateReadReply>>(),
|
||||
handler:
|
||||
vi.fn<
|
||||
(
|
||||
input: unknown,
|
||||
channel?: WorkerTaskChannel,
|
||||
control?: WorkerTaskControl,
|
||||
) => Promise<OpenClawStateReadReply>
|
||||
>(),
|
||||
control: {
|
||||
runNativeSection: async <T>(operation: () => T | Promise<T>) => operation(),
|
||||
throwIfCancelled: () => {},
|
||||
} satisfies WorkerTaskControl,
|
||||
open: vi.fn(),
|
||||
admit: vi.fn<() => void>(),
|
||||
query: vi.fn<() => []>(),
|
||||
settle: vi.fn<(operation: (source: { db: object }) => unknown) => unknown>(),
|
||||
|
|
@ -48,7 +65,11 @@ vi.mock("node:worker_threads", async (importOriginal) => {
|
|||
});
|
||||
vi.mock("../infra/worker-task-server.js", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("../infra/worker-task-server.js")>()),
|
||||
serveOwnedWorkerTasks: mock.handler.mockImplementation.bind(mock.handler),
|
||||
serveOwnedWorkerTasks: (handler: WorkerTaskHandler<OpenClawStateReadReply>) => {
|
||||
mock.handler.mockImplementation(async (input, channel, control) =>
|
||||
handler(input, channel, control ?? mock.control),
|
||||
);
|
||||
},
|
||||
}));
|
||||
vi.mock("./backup-run-records.kernel.js", async (importOriginal) => {
|
||||
mock.unrelatedHandlerLoaded = true;
|
||||
|
|
@ -88,6 +109,7 @@ vi.mock("../infra/gateway-owner-lease.worker.js", async (importOriginal) => ({
|
|||
// mock-isolation: Dispatch tests supply fixture handles without opening their synthetic source paths.
|
||||
vi.mock("./openclaw-state-db-read-connection.js", () => ({
|
||||
closeRetainedOpenClawStateReadConnections: vi.fn(),
|
||||
openOpenClawStateReadConnection: mock.open,
|
||||
readOpenClawStateReadOnlyLocation: mock.settle,
|
||||
withOpenClawStateReadOnlyLocation: (operation: (source: { db: object }) => unknown) => {
|
||||
mock.admit();
|
||||
|
|
@ -111,6 +133,9 @@ afterEach(async () => {
|
|||
});
|
||||
|
||||
beforeEach(() => {
|
||||
mock.open.mockReset().mockImplementation(() => {
|
||||
throw new Error("Unexpected database open");
|
||||
});
|
||||
mock.database = undefined;
|
||||
mock.admit.mockReset();
|
||||
mock.query.mockReset().mockReturnValue([]);
|
||||
|
|
@ -256,3 +281,188 @@ it.each([
|
|||
}
|
||||
expect(mock.query).toHaveBeenCalledTimes(outcome === "schema-error" ? 0 : 1);
|
||||
});
|
||||
|
||||
describe.each([
|
||||
{ type: "subagents.restore" },
|
||||
{
|
||||
type: "meetingTranscripts.export",
|
||||
format: "library",
|
||||
selector: "fixture-session",
|
||||
includeNotes: true,
|
||||
},
|
||||
] as const)("$type stream loading", (command) => {
|
||||
const loaded = new Set<string>();
|
||||
const freshAdmission = vi.fn<() => void>();
|
||||
const nativeSection = vi.fn();
|
||||
const restore =
|
||||
vi.fn<
|
||||
typeof import("../agents/subagents/registry/subagent-registry.store.restore.worker.js").streamSubagentRegistryInWorker
|
||||
>();
|
||||
const exportTranscript =
|
||||
vi.fn<typeof import("../transcripts/store-export.worker.js").streamTranscriptExportInWorker>();
|
||||
const channel: WorkerTaskChannel = {
|
||||
consumeInput: vi.fn(),
|
||||
notify: vi.fn(),
|
||||
request: vi.fn<WorkerTaskChannel["request"]>(),
|
||||
};
|
||||
const control: WorkerTaskControl = {
|
||||
runNativeSection: async <T>(operation: () => T | Promise<T>) => {
|
||||
nativeSection();
|
||||
return operation();
|
||||
},
|
||||
throwIfCancelled: () => {},
|
||||
};
|
||||
let holdImport = false;
|
||||
let importStarted: PromiseWithResolvers<void>;
|
||||
let releaseImport: PromiseWithResolvers<void>;
|
||||
|
||||
beforeEach(async () => {
|
||||
// Each command must start cold, including Vitest's separate mock-module cache.
|
||||
vi.resetModules();
|
||||
loaded.clear();
|
||||
holdImport = false;
|
||||
importStarted = Promise.withResolvers<void>();
|
||||
releaseImport = Promise.withResolvers<void>();
|
||||
freshAdmission.mockReset();
|
||||
nativeSection.mockReset();
|
||||
restore.mockReset().mockImplementation(async (_input, _channel, onAdmitted) => {
|
||||
onAdmitted();
|
||||
return 2;
|
||||
});
|
||||
exportTranscript
|
||||
.mockReset()
|
||||
.mockImplementation(async (_input, _command, _channel, onAdmitted) => {
|
||||
onAdmitted();
|
||||
return { ok: true };
|
||||
});
|
||||
vi.doMock("./openclaw-state-read-handlers.js", async (importOriginal) => {
|
||||
loaded.add("catchall");
|
||||
return importOriginal<typeof import("./openclaw-state-read-handlers.js")>();
|
||||
});
|
||||
vi.doMock(
|
||||
"../agents/subagents/registry/subagent-registry.store.worker.js",
|
||||
async (importOriginal) => {
|
||||
loaded.add("mixed-store");
|
||||
return importOriginal<
|
||||
typeof import("../agents/subagents/registry/subagent-registry.store.worker.js")
|
||||
>();
|
||||
},
|
||||
);
|
||||
vi.doMock(
|
||||
"../agents/subagents/registry/subagent-registry.store.restore.worker.js",
|
||||
async (importOriginal) => {
|
||||
loaded.add("subagents.restore");
|
||||
if (holdImport && command.type === "subagents.restore") {
|
||||
importStarted.resolve();
|
||||
await releaseImport.promise;
|
||||
}
|
||||
return {
|
||||
...(await importOriginal<
|
||||
typeof import("../agents/subagents/registry/subagent-registry.store.restore.worker.js")
|
||||
>()),
|
||||
streamSubagentRegistryInWorker: restore,
|
||||
};
|
||||
},
|
||||
);
|
||||
vi.doMock("../transcripts/store-export.worker.js", async (importOriginal) => {
|
||||
loaded.add("meetingTranscripts.export");
|
||||
if (holdImport && command.type === "meetingTranscripts.export") {
|
||||
importStarted.resolve();
|
||||
await releaseImport.promise;
|
||||
}
|
||||
return {
|
||||
...(await importOriginal<typeof import("../transcripts/store-export.worker.js")>()),
|
||||
streamTranscriptExportInWorker: exportTranscript,
|
||||
};
|
||||
});
|
||||
const { openClawStateDatabaseCache } = await import("./openclaw-state-db-cache.js");
|
||||
vi.spyOn(
|
||||
openClawStateDatabaseCache,
|
||||
"assertOpenClawStateDatabaseFreshOpenAllowedAtPath",
|
||||
).mockImplementation(freshAdmission);
|
||||
await import("./openclaw-state-read.worker.js");
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
releaseImport.resolve();
|
||||
vi.doUnmock("./openclaw-state-read-handlers.js");
|
||||
vi.doUnmock("../agents/subagents/registry/subagent-registry.store.worker.js");
|
||||
vi.doUnmock("../agents/subagents/registry/subagent-registry.store.restore.worker.js");
|
||||
vi.doUnmock("../transcripts/store-export.worker.js");
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("loads only its stream owner and preserves the reply", async () => {
|
||||
const input = { ...request, checkFreshAdmission: true, command };
|
||||
const reply = await mock.handler(input, channel, control);
|
||||
expect(reply).toEqual(
|
||||
command.type === "subagents.restore"
|
||||
? { ok: true, type: command.type, sourceAdmitted: true, count: 2 }
|
||||
: { ok: true, type: command.type, sourceAdmitted: true, result: { ok: true } },
|
||||
);
|
||||
if (command.type === "subagents.restore") {
|
||||
expect(restore).toHaveBeenCalledExactlyOnceWith(input, channel, expect.any(Function));
|
||||
expect(exportTranscript).not.toHaveBeenCalled();
|
||||
} else {
|
||||
expect(exportTranscript).toHaveBeenCalledExactlyOnceWith(
|
||||
input,
|
||||
command,
|
||||
channel,
|
||||
expect.any(Function),
|
||||
);
|
||||
expect(restore).not.toHaveBeenCalled();
|
||||
}
|
||||
expect(freshAdmission).toHaveBeenCalledOnce();
|
||||
expect(nativeSection).toHaveBeenCalledOnce();
|
||||
expect(loaded).toEqual(new Set([command.type]));
|
||||
});
|
||||
|
||||
it("requires a receiver before entering either stream owner", async () => {
|
||||
const reply = await mock.handler({ ...request, command }, undefined, control);
|
||||
const operation =
|
||||
command.type === "subagents.restore" ? "Subagent restore" : "Transcript export";
|
||||
expect(reply).toMatchObject({
|
||||
ok: false,
|
||||
message: `${operation} requires a bounded receiver`,
|
||||
sourceAdmitted: undefined,
|
||||
});
|
||||
expect(restore).not.toHaveBeenCalled();
|
||||
expect(exportTranscript).not.toHaveBeenCalled();
|
||||
expect(mock.open).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("checks fresh admission after the pending owner import settles", async () => {
|
||||
holdImport = true;
|
||||
const pending = mock.handler(
|
||||
{ ...request, checkFreshAdmission: true, command },
|
||||
channel,
|
||||
control,
|
||||
);
|
||||
let reply: OpenClawStateReadReply | undefined;
|
||||
try {
|
||||
await Promise.race([
|
||||
importStarted.promise,
|
||||
pending.then(() => {
|
||||
throw new Error("Worker returned before loading the selected stream owner");
|
||||
}),
|
||||
]);
|
||||
expect(freshAdmission).not.toHaveBeenCalled();
|
||||
freshAdmission.mockImplementation(() => {
|
||||
throw new Error("Fresh source admission refused");
|
||||
});
|
||||
} finally {
|
||||
releaseImport.resolve();
|
||||
reply = await pending;
|
||||
}
|
||||
expect(reply).toMatchObject({
|
||||
ok: false,
|
||||
message: "Fresh source admission refused",
|
||||
sourceAdmitted: undefined,
|
||||
});
|
||||
expect(freshAdmission).toHaveBeenCalledOnce();
|
||||
expect(nativeSection).not.toHaveBeenCalled();
|
||||
expect(restore).not.toHaveBeenCalled();
|
||||
expect(exportTranscript).not.toHaveBeenCalled();
|
||||
expect(mock.open).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import {
|
|||
} from "./openclaw-state-db-read-connection.js";
|
||||
import { postgresStateReadSnapshot } from "./openclaw-state-postgres-snapshot-context.js";
|
||||
import { stateReadRegistry } from "./openclaw-state-read-operation-registry.js";
|
||||
import { prepareStateStream } from "./openclaw-state-read-stream.js";
|
||||
import type {
|
||||
OpenClawStateReadReply,
|
||||
OpenClawStateReadRequest,
|
||||
|
|
@ -49,7 +50,16 @@ const read: WorkerTaskHandler<OpenClawStateReadReply> = async function read(
|
|||
input.command.type === "localWorkspace.manifest"
|
||||
? await import("../gateway/worker-environments/local-workspace-manifest-scan.js")
|
||||
: undefined;
|
||||
// Finish the selected owner's import before fresh admission, not during stream execution.
|
||||
const readStream =
|
||||
input.command.type === "subagents.restore" ||
|
||||
input.command.type === "meetingTranscripts.export"
|
||||
? await prepareStateStream(input, input.command, channel, () => {
|
||||
sourceAdmitted = true;
|
||||
})
|
||||
: undefined;
|
||||
const dedicatedRead =
|
||||
readStream !== undefined ||
|
||||
input.command.type === "admit" ||
|
||||
input.command.type === "plugins.metadata.read" ||
|
||||
input.command.type === "gatewayOwnerLease.read" ||
|
||||
|
|
@ -191,16 +201,15 @@ const read: WorkerTaskHandler<OpenClawStateReadReply> = async function read(
|
|||
: { status: "unavailable" },
|
||||
};
|
||||
}
|
||||
if (command.type === "subagents.restore" || command.type === "meetingTranscripts.export") {
|
||||
if (!readStream) {
|
||||
throw new Error("State read stream is unavailable");
|
||||
}
|
||||
return await control.runNativeSection(readStream);
|
||||
}
|
||||
if (!handlers) {
|
||||
throw new Error("State read command handlers are unavailable");
|
||||
}
|
||||
if (command.type === "subagents.restore" || command.type === "meetingTranscripts.export") {
|
||||
return await control.runNativeSection(() =>
|
||||
handlers.readStateStream(input, command, channel, () => {
|
||||
sourceAdmitted = true;
|
||||
}),
|
||||
);
|
||||
}
|
||||
if (command.type === "subagents.sessionList") {
|
||||
const result = readOpenClawStateReadOnlyLocation(
|
||||
({ db }) => {
|
||||
|
|
|
|||
|
|
@ -2,15 +2,23 @@ import { execFileSync } from "node:child_process";
|
|||
import fs from "node:fs";
|
||||
import { createRequire } from "node:module";
|
||||
import path from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { build } from "tsdown";
|
||||
import { expect, it } from "vitest";
|
||||
import { hasUnjoinedWork, runManagedCommand } from "../../scripts/lib/managed-child-process.mts";
|
||||
import buildConfigs from "../../tsdown.config.ts";
|
||||
import { createTempDirTracker } from "../helpers/temp-dir.js";
|
||||
import { createScriptTestHarness } from "./test-helpers.js";
|
||||
|
||||
const configs = Array.isArray(buildConfigs) ? buildConfigs : [buildConfigs];
|
||||
const { createTempDir } = createScriptTestHarness();
|
||||
const handlersModule = "/src/state/openclaw-state-read-handlers.ts";
|
||||
const registryModule = "/src/state/openclaw-agent-db-registry.read.ts";
|
||||
const mixedStoreModule = "/src/agents/subagents/registry/subagent-registry.store.worker.ts";
|
||||
const streamModules = [
|
||||
"/src/agents/subagents/registry/subagent-registry.store.restore.worker.ts",
|
||||
"/src/transcripts/store-export.worker.ts",
|
||||
];
|
||||
const hasModule = (ids: Iterable<string>, ...suffixes: string[]) =>
|
||||
[...ids].some((id) => suffixes.some((suffix) => id.endsWith(suffix)));
|
||||
|
||||
|
|
@ -48,78 +56,208 @@ it("preserves installed native and asset locations when filesystem helpers are b
|
|||
it.each([
|
||||
["state/openclaw-state-read.worker", ".js"],
|
||||
["worker/openclaw-state-read.worker", ".mjs"],
|
||||
])("keeps unrelated handlers out of %s startup", async (entryName, extension) => {
|
||||
const selected = configs.find(
|
||||
(config) =>
|
||||
typeof config.entry === "object" &&
|
||||
config.entry !== null &&
|
||||
Object.hasOwn(config.entry, entryName),
|
||||
);
|
||||
if (!selected) {
|
||||
throw new Error(`Missing state read worker build config: ${entryName}`);
|
||||
}
|
||||
const { bundles } = await build({
|
||||
...selected,
|
||||
config: false,
|
||||
outDir: fs.realpathSync(createTempDir("openclaw-state-read-build-")),
|
||||
dts: false,
|
||||
logLevel: "silent",
|
||||
});
|
||||
try {
|
||||
const chunks = new Map(
|
||||
bundles.flatMap((bundle) =>
|
||||
bundle.chunks
|
||||
.filter((chunk) => chunk.type === "chunk")
|
||||
.map((chunk) => [chunk.fileName, chunk] as const),
|
||||
),
|
||||
])(
|
||||
"keeps unrelated handlers out of %s startup",
|
||||
async (entryName, extension) => {
|
||||
const selected = configs.find(
|
||||
(config) =>
|
||||
typeof config.entry === "object" &&
|
||||
config.entry !== null &&
|
||||
Object.hasOwn(config.entry, entryName),
|
||||
);
|
||||
const resolveImport = (name: string, specifier: string) =>
|
||||
specifier.startsWith(".")
|
||||
? path.posix.normalize(path.posix.join(path.posix.dirname(name), specifier))
|
||||
: specifier;
|
||||
const renderedModules = (name: string) =>
|
||||
Object.entries(chunks.get(name)?.modules ?? {})
|
||||
.filter(([, module]) => module.renderedLength > 0)
|
||||
.map(([id]) => id.replaceAll("\\", "/"));
|
||||
const staticClosure = (entryFile: string) => {
|
||||
const queue = new Set([entryFile]);
|
||||
const modules = new Set<string>();
|
||||
const dynamicImports = new Set<string>();
|
||||
for (const name of queue) {
|
||||
const chunk = chunks.get(name);
|
||||
if (!chunk) {
|
||||
throw new Error(`Missing state read worker chunk: ${name}`);
|
||||
}
|
||||
renderedModules(name).forEach((id) => modules.add(id));
|
||||
chunk.dynamicImports.forEach((specifier) =>
|
||||
dynamicImports.add(resolveImport(name, specifier)),
|
||||
);
|
||||
for (const specifier of chunk.imports) {
|
||||
const target = resolveImport(name, specifier);
|
||||
if (chunks.has(target)) {
|
||||
queue.add(target);
|
||||
if (!selected) {
|
||||
throw new Error(`Missing state read worker build config: ${entryName}`);
|
||||
}
|
||||
const buildDirs = createTempDirTracker();
|
||||
const outDir = fs.realpathSync(buildDirs.make("openclaw-state-read-build-"));
|
||||
let bundles: Awaited<ReturnType<typeof build>>["bundles"] = [];
|
||||
let childrenJoined = true;
|
||||
try {
|
||||
({ bundles } = await build({
|
||||
...selected,
|
||||
config: false,
|
||||
outDir,
|
||||
dts: false,
|
||||
logLevel: "silent",
|
||||
}));
|
||||
const chunks = new Map(
|
||||
bundles.flatMap((bundle) =>
|
||||
bundle.chunks
|
||||
.filter((chunk) => chunk.type === "chunk")
|
||||
.map((chunk) => [chunk.fileName, chunk] as const),
|
||||
),
|
||||
);
|
||||
const resolveImport = (name: string, specifier: string) =>
|
||||
specifier.startsWith(".")
|
||||
? path.posix.normalize(path.posix.join(path.posix.dirname(name), specifier))
|
||||
: specifier;
|
||||
const renderedModules = (name: string) =>
|
||||
Object.entries(chunks.get(name)?.modules ?? {})
|
||||
.filter(([, module]) => module.renderedLength > 0)
|
||||
.map(([id]) => id.replaceAll("\\", "/"));
|
||||
const staticClosure = (entryFile: string) => {
|
||||
const queue = new Set([entryFile]);
|
||||
const modules = new Set<string>();
|
||||
const dynamicImports = new Set<string>();
|
||||
for (const name of queue) {
|
||||
const chunk = chunks.get(name);
|
||||
if (!chunk) {
|
||||
throw new Error(`Missing state read worker chunk: ${name}`);
|
||||
}
|
||||
renderedModules(name).forEach((id) => modules.add(id));
|
||||
chunk.dynamicImports.forEach((specifier) =>
|
||||
dynamicImports.add(resolveImport(name, specifier)),
|
||||
);
|
||||
for (const specifier of chunk.imports) {
|
||||
const target = resolveImport(name, specifier);
|
||||
if (chunks.has(target)) {
|
||||
queue.add(target);
|
||||
}
|
||||
}
|
||||
}
|
||||
return { modules, dynamicImports };
|
||||
};
|
||||
const startup = staticClosure(`${entryName}${extension}`);
|
||||
expect(hasModule(startup.modules, handlersModule)).toBe(false);
|
||||
expect(hasModule([...chunks.keys()].flatMap(renderedModules), handlersModule)).toBe(true);
|
||||
if (entryName === "state/openclaw-state-read.worker") {
|
||||
const registryTargets = [...startup.dynamicImports].filter((name) =>
|
||||
hasModule(renderedModules(name), registryModule),
|
||||
);
|
||||
expect(registryTargets).toHaveLength(1);
|
||||
const registry = staticClosure(registryTargets[0]!);
|
||||
expect(hasModule(registry.modules, registryModule)).toBe(true);
|
||||
expect(
|
||||
hasModule(registry.modules, handlersModule, "/src/state/backup-run-records.kernel.ts"),
|
||||
).toBe(false);
|
||||
}
|
||||
return { modules, dynamicImports };
|
||||
};
|
||||
const startup = staticClosure(`${entryName}${extension}`);
|
||||
expect(hasModule(startup.modules, handlersModule)).toBe(false);
|
||||
expect(hasModule([...chunks.keys()].flatMap(renderedModules), handlersModule)).toBe(true);
|
||||
if (entryName === "state/openclaw-state-read.worker") {
|
||||
const registryTargets = [...startup.dynamicImports].filter((name) =>
|
||||
hasModule(renderedModules(name), registryModule),
|
||||
);
|
||||
expect(registryTargets).toHaveLength(1);
|
||||
const registry = staticClosure(registryTargets[0]!);
|
||||
expect(hasModule(registry.modules, registryModule)).toBe(true);
|
||||
expect(
|
||||
hasModule(registry.modules, handlersModule, "/src/state/backup-run-records.kernel.ts"),
|
||||
).toBe(false);
|
||||
[...startup.modules].filter((id) =>
|
||||
[mixedStoreModule, ...streamModules].some((module) => id.endsWith(module)),
|
||||
),
|
||||
).toEqual([]);
|
||||
for (const streamModule of streamModules) {
|
||||
// Manual chunks identify contained modules; facades identify an owner
|
||||
// whose rendered body can live in a static dependency.
|
||||
const streamTargets = [...startup.dynamicImports].filter((name) => {
|
||||
const chunk = chunks.get(name);
|
||||
return (
|
||||
chunk?.facadeModuleId?.replaceAll("\\", "/").endsWith(streamModule) === true ||
|
||||
chunk?.moduleIds.some((id) => id.replaceAll("\\", "/").endsWith(streamModule)) === true
|
||||
);
|
||||
});
|
||||
const [streamTarget] = streamTargets;
|
||||
if (streamTargets.length !== 1 || streamTarget === undefined) {
|
||||
throw new Error(
|
||||
`Expected exactly one entry-selected stream chunk for ${streamModule}, found ${streamTargets.length}: ${entryName}`,
|
||||
);
|
||||
}
|
||||
const stream = staticClosure(streamTarget);
|
||||
expect(hasModule(stream.modules, streamModule)).toBe(true);
|
||||
const unrelatedModules = [
|
||||
handlersModule,
|
||||
mixedStoreModule,
|
||||
...streamModules.filter((module) => module !== streamModule),
|
||||
];
|
||||
expect(
|
||||
[...stream.modules].filter((id) =>
|
||||
unrelatedModules.some((module) => id.endsWith(module)),
|
||||
),
|
||||
`${entryName}: ${streamModule}`,
|
||||
).toEqual([]);
|
||||
}
|
||||
if (entryName === "worker/openclaw-state-read.worker") {
|
||||
const workerUrl = pathToFileURL(path.join(outDir, `${entryName}${extension}`)).href;
|
||||
// Each cold process must initialize its selected owner through the emitted continuation.
|
||||
for (const mode of ["restore", "export"] as const) {
|
||||
const stdout: Buffer[] = [];
|
||||
const stderr: Buffer[] = [];
|
||||
let stdoutBytes = 0;
|
||||
let stderrBytes = 0;
|
||||
const closed = new Set<string>();
|
||||
const required = [
|
||||
"exit",
|
||||
"close",
|
||||
"stdout-end",
|
||||
"stdout-close",
|
||||
"stderr-end",
|
||||
"stderr-close",
|
||||
];
|
||||
childrenJoined = false;
|
||||
let code: number;
|
||||
try {
|
||||
code = await runManagedCommand({
|
||||
bin: process.execPath,
|
||||
args: [
|
||||
"--import",
|
||||
path.resolve("scripts/tsx.mjs"),
|
||||
path.resolve("test/scripts/state-read-worker-stream.test-support.ts"),
|
||||
mode,
|
||||
workerUrl,
|
||||
],
|
||||
cwd: process.cwd(),
|
||||
env: process.env,
|
||||
shell: false,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
timeoutMs: 60_000,
|
||||
timeoutKillGraceMs: 10_000,
|
||||
signalKillGraceMs: 10_000,
|
||||
abortKillGraceMs: 10_000,
|
||||
cleanupDrainTimeoutMs: 5_000,
|
||||
timeoutForceKillOnLeaderExit: true,
|
||||
requireProcessTreeExit: process.platform !== "win32",
|
||||
waitForCleanupRelease: true,
|
||||
onReady(child) {
|
||||
child.stdout!.on("data", (bytes: Buffer) => {
|
||||
stdoutBytes += bytes.length;
|
||||
if (stdoutBytes <= 1024 * 1024) {
|
||||
stdout.push(bytes);
|
||||
}
|
||||
});
|
||||
child.stderr!.on("data", (bytes: Buffer) => {
|
||||
stderrBytes += bytes.length;
|
||||
if (stderrBytes <= 1024 * 1024) {
|
||||
stderr.push(bytes);
|
||||
}
|
||||
});
|
||||
child.once("exit", () => closed.add("exit"));
|
||||
child.once("close", () => closed.add("close"));
|
||||
child.stdout!.once("end", () => closed.add("stdout-end"));
|
||||
child.stdout!.once("close", () => closed.add("stdout-close"));
|
||||
child.stderr!.once("end", () => closed.add("stderr-end"));
|
||||
child.stderr!.once("close", () => closed.add("stderr-close"));
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
childrenJoined =
|
||||
!hasUnjoinedWork(error) && required.every((event) => closed.has(event));
|
||||
throw error;
|
||||
}
|
||||
childrenJoined = required.every((event) => closed.has(event));
|
||||
expect(childrenJoined, `${mode}: child and streams joined`).toBe(true);
|
||||
expect(stdoutBytes).toBeLessThanOrEqual(1024 * 1024);
|
||||
expect(stderrBytes).toBeLessThanOrEqual(1024 * 1024);
|
||||
expect(code, Buffer.concat(stderr).toString("utf8")).toBe(0);
|
||||
expect(JSON.parse(Buffer.concat(stdout).toString("utf8"))).toEqual({
|
||||
mode,
|
||||
workerUrl,
|
||||
values:
|
||||
mode === "restore"
|
||||
? ["restore-first", "restore-second", "restore-third"]
|
||||
: ["first", "second"],
|
||||
});
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (childrenJoined) {
|
||||
for (const bundle of bundles) {
|
||||
await bundle[Symbol.asyncDispose]();
|
||||
}
|
||||
buildDirs.cleanup();
|
||||
} else {
|
||||
console.error(`State read worker build retained for unjoined fixture: ${outDir}`);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
for (const bundle of bundles) {
|
||||
await bundle[Symbol.asyncDispose]();
|
||||
}
|
||||
}
|
||||
});
|
||||
},
|
||||
180_000,
|
||||
);
|
||||
|
|
|
|||
106
test/scripts/state-read-worker-stream.test-support.ts
Normal file
106
test/scripts/state-read-worker-stream.test-support.ts
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
import assert from "node:assert/strict";
|
||||
import {
|
||||
createSubagentStoreRunFixture,
|
||||
saveSubagentRegistryToSqlite,
|
||||
} from "../../src/agents/subagents/registry/subagent-registry-state.fixture.test-support.js";
|
||||
import { readAllSubagentRunsInWorker } from "../../src/agents/subagents/registry/subagent-registry.store.read.js";
|
||||
import {
|
||||
registerSealedRuntimeProcessEntrypoint,
|
||||
resolveRuntimeProcessEntrypointUrl,
|
||||
} from "../../src/infra/runtime-process-url.js";
|
||||
import { captureRuntimeWorkerSource } from "../../src/infra/runtime-worker-generation.js";
|
||||
import {
|
||||
captureRetainedNativeWorkerSource,
|
||||
closeDefaultRetainedNativeWorkerSource,
|
||||
} from "../../src/infra/worker-native-lifecycle.js";
|
||||
import { closeOpenClawStateDatabaseAsync } from "../../src/state/openclaw-state-db.js";
|
||||
import { captureOpenClawStateWorkerContext } from "../../src/state/openclaw-state-worker-context.js";
|
||||
import { createOpenClawTestState } from "../../src/test-utils/openclaw-test-state.js";
|
||||
import { transcriptSessionSelector, TranscriptsStore } from "../../src/transcripts/store.js";
|
||||
|
||||
const [mode, entrypoint] = process.argv.slice(2);
|
||||
assert.ok(mode === "restore" || mode === "export", "Expected restore or export fixture");
|
||||
assert.ok(entrypoint, "Expected emitted state read entrypoint");
|
||||
const emitted = new URL(entrypoint);
|
||||
assert.equal(emitted.protocol, "file:");
|
||||
|
||||
async function closeRuntime(): Promise<void> {
|
||||
try {
|
||||
await closeOpenClawStateDatabaseAsync();
|
||||
} finally {
|
||||
await closeDefaultRetainedNativeWorkerSource();
|
||||
}
|
||||
}
|
||||
|
||||
const state = await createOpenClawTestState({ scenario: "minimal", applyEnv: true });
|
||||
let values: string[];
|
||||
let workerUrl: string;
|
||||
try {
|
||||
const store = new TranscriptsStore(state.statePath("transcripts"), { env: state.env });
|
||||
const session = {
|
||||
sessionId: "emitted-stream",
|
||||
source: { providerId: "manual-transcript" },
|
||||
startedAt: "2026-07-01T10:00:00.000Z",
|
||||
};
|
||||
if (mode === "restore") {
|
||||
const rows = [
|
||||
["restore-third", 30],
|
||||
["restore-first", 10],
|
||||
["restore-second", 20],
|
||||
] as const;
|
||||
saveSubagentRegistryToSqlite(
|
||||
new Map(
|
||||
rows.map(([runId, createdAt]) => [
|
||||
runId,
|
||||
createSubagentStoreRunFixture({
|
||||
runId,
|
||||
childSessionKey: `agent:main:subagent:${runId}`,
|
||||
task: runId,
|
||||
createdAt,
|
||||
completion: { required: false },
|
||||
delivery: { status: "not_required" },
|
||||
}),
|
||||
]),
|
||||
),
|
||||
);
|
||||
} else {
|
||||
await store.writeSession(session);
|
||||
await store.appendUtteranceForSession(session, { text: "first" });
|
||||
await store.appendUtteranceForSession(session, { text: "second" });
|
||||
}
|
||||
|
||||
// Seeding may capture source workers. Retire them before the first emitted read capture.
|
||||
await closeRuntime();
|
||||
registerSealedRuntimeProcessEntrypoint("stateRead", emitted);
|
||||
const nativeSource = captureRetainedNativeWorkerSource();
|
||||
assert.equal(nativeSource.hasActiveWorkers, false);
|
||||
const captured = captureRuntimeWorkerSource(resolveRuntimeProcessEntrypointUrl("stateRead"));
|
||||
workerUrl = captured.moduleUrl.href;
|
||||
assert.equal(workerUrl, emitted.href);
|
||||
assert.equal(captured.runtimeGeneration, undefined);
|
||||
|
||||
if (mode === "restore") {
|
||||
const runs = await readAllSubagentRunsInWorker(
|
||||
captureOpenClawStateWorkerContext({ env: state.env }),
|
||||
);
|
||||
values = [...runs.keys()];
|
||||
assert.deepEqual(values, ["restore-first", "restore-second", "restore-third"]);
|
||||
assert.deepEqual(
|
||||
[...runs.values()].map((row) => row.task),
|
||||
values,
|
||||
);
|
||||
} else {
|
||||
values = [];
|
||||
const result = await store.streamExport(transcriptSessionSelector(session), false, (rows) => {
|
||||
values.push(...rows.map((row) => row.text));
|
||||
});
|
||||
assert.deepEqual(values, ["first", "second"]);
|
||||
assert.equal(result.entry.session.sessionId, session.sessionId);
|
||||
assert.equal(result.entry.session.startedAt, session.startedAt);
|
||||
assert.equal(result.notes, undefined);
|
||||
}
|
||||
} finally {
|
||||
await closeRuntime();
|
||||
await state.cleanup();
|
||||
}
|
||||
console.log(JSON.stringify({ mode, workerUrl, values }));
|
||||
|
|
@ -54,6 +54,7 @@ import {
|
|||
} from "./scripts/lib/worker-deploy-build-plugin.mts";
|
||||
import { buildPackageDistEntriesFromExports } from "./scripts/lib/workspace-package-entries.mts";
|
||||
|
||||
type OutputOptions = Exclude<NonNullable<UserConfig["outputOptions"]>, Function>;
|
||||
type InputOptionsFactory = Extract<NonNullable<UserConfig["inputOptions"]>, Function>;
|
||||
type InputOptionsArg = InputOptionsFactory extends (
|
||||
options: infer Options,
|
||||
|
|
@ -264,7 +265,10 @@ function nodeBuildConfig(
|
|||
};
|
||||
}
|
||||
|
||||
function workerDeployBuildConfig(entry: Record<string, string>, split = false): UserConfig {
|
||||
function workerDeployBuildConfig(
|
||||
entry: Record<string, string>,
|
||||
split: OutputOptions["codeSplitting"] = false,
|
||||
): UserConfig {
|
||||
return {
|
||||
name: TSDOWN_UNIFIED_CONFIG_GROUP,
|
||||
entry,
|
||||
|
|
@ -1032,7 +1036,31 @@ const configs: UserConfig[] = [
|
|||
}),
|
||||
workerDeployBuildConfig(
|
||||
{ "worker/openclaw-state-read.worker": "src/worker/worker-deploy-state-read.ts" },
|
||||
true,
|
||||
{
|
||||
// Keep lazy owners separate without capturing their shared dependencies.
|
||||
groups: [
|
||||
{
|
||||
name: "state-read-handlers",
|
||||
test: /(?:^|[\\/])src[\\/]state[\\/]openclaw-state-read-handlers\.ts$/u,
|
||||
includeDependenciesRecursively: false,
|
||||
},
|
||||
{
|
||||
name: "state-read-subagent-store",
|
||||
test: /(?:^|[\\/])src[\\/]agents[\\/]subagents[\\/]registry[\\/]subagent-registry\.store\.worker\.ts$/u,
|
||||
includeDependenciesRecursively: false,
|
||||
},
|
||||
{
|
||||
name: "state-read-subagent-restore",
|
||||
test: /(?:^|[\\/])src[\\/]agents[\\/]subagents[\\/]registry[\\/]subagent-registry\.store\.restore\.worker\.ts$/u,
|
||||
includeDependenciesRecursively: false,
|
||||
},
|
||||
{
|
||||
name: "state-read-transcript-export",
|
||||
test: /(?:^|[\\/])src[\\/]transcripts[\\/]store-export\.worker\.ts$/u,
|
||||
includeDependenciesRecursively: false,
|
||||
},
|
||||
],
|
||||
},
|
||||
),
|
||||
workerDeployBuildConfig({
|
||||
"worker/worker-native-lifecycle.worker": "src/infra/worker-native-lifecycle.worker.ts",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue