fix(sqlite): retain completed worker evidence and unresolved cleanup ownership

This commit is contained in:
morluto committed 2026-10-14 09:47:27 +08:00
1 parent 31b99306aa
commit b940ad0d65
4 files changed
+198 -34

No files matched your search

+10 -3
View File
@@ -1,6 +1,7 @@
import type {
OwnedProviderProcessSpawnOptions,
ProviderProcessSnapshot,
ProviderProcessStopResult,
SpawnedOwnedProviderProcess,
} from "./ProviderProcess.js";
import {
@@ -63,7 +64,10 @@ export const runOwnedCommand = async (
options: {
readonly signal?: AbortSignal | undefined;
/** Synchronously accept unresolved cleanup ownership; execution still returns its original outcome. Must not throw. */
readonly retainCleanup?: (owner: ProviderProcessSupervisor) => void;
readonly retainCleanup?: (
owner: ProviderProcessSupervisor,
result: Extract<ProviderProcessStopResult, { status: "incomplete" }>,
) => void;
readonly launcher?: (
input: OwnedProviderProcessSpawnOptions,
) => Promise<SpawnedOwnedProviderProcess>;
@@ -102,7 +106,10 @@ const collectOwnedCommand = async (
) => Promise<SpawnedOwnedProviderProcess>,
deadline: ProviderStartupDeadline,
callerSignal?: AbortSignal,
retainCleanup?: (owner: ProviderProcessSupervisor) => void,
retainCleanup?: (
owner: ProviderProcessSupervisor,
result: Extract<ProviderProcessStopResult, { status: "incomplete" }>,
) => void,
): Promise<ProviderProcessSnapshot> => {
const signal = deadline.signal;
let launched: SpawnedOwnedProviderProcess;
@@ -216,7 +223,7 @@ const collectOwnedCommand = async (
const stopped = await supervisor.stop();
const snapshot = supervisor.snapshot();
if (stopped.status === "incomplete" && retainCleanup !== undefined)
retainCleanup(supervisor);
retainCleanup(supervisor, stopped);
else if (stopped.status === "incomplete")
throw new OwnedCommandFailure(
failure?.reason ?? "process",
+92 -31
View File
@@ -17,6 +17,7 @@ import {
AnalysisResourceConstraintError,
} from "../domain/analysisErrorCore.js";
import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import type { JsonValue } from "../domain/jsonValue.js";
import { createEvidence } from "../domain/evidence.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
import { err, ok, type Result } from "../domain/result.js";
@@ -29,8 +30,14 @@ import {
} from "../domain/sqlite/sqliteDatabase.js";
import { isAbsolute } from "node:path";
import { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js";
import { OwnedCommandFailure } from "../process/OwnedCommand.js";
import type { ProviderProcessSupervisor } from "../process/ProviderProcess.js";
import {
OwnedCommandFailure,
ownedCommandResources,
} from "../process/OwnedCommand.js";
import type {
ProviderProcessStopResult,
ProviderProcessSupervisor,
} from "../process/ProviderProcess.js";
import { sqliteDatabaseFailure } from "./SqliteDatabaseFailures.js";
import { captureSqliteDatabaseSnapshot } from "./SqliteDatabaseSnapshot.js";
import { SQLITE_DATABASE_LIMITS } from "./SqliteDatabaseLimits.js";
@@ -228,6 +235,9 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
): Promise<SqliteOutcome> {
let root: Pick<PrivateRuntimeRoot, "path" | "close"> | undefined;
let processOwner: ProviderProcessSupervisor | undefined;
let processCleanup:
| Extract<ProviderProcessStopResult, { status: "incomplete" }>
| undefined;
let retainedOutput: AnalysisCapturedOutput | undefined;
let phase = "configuration";
let result: Result<AnalysisExecution, AnalysisError>;
@@ -258,6 +268,10 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
captured: (output) => {
retainedOutput = output;
},
retainCleanup: (owner, result) => {
processOwner = owner;
processCleanup = result;
},
});
const validated = sqliteDatabaseSchema.safeParse({
artifact: snapshot.artifact,
@@ -293,6 +307,39 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
output: retainedOutput,
process: processOwner,
});
if (processCleanup !== undefined) {
const ownership = processOwner.launch.ownership;
return err(
sqliteCleanupFailure(
ownership === undefined
? ["owned-sqlite-worker", root.path]
: ownedCommandResources(ownership, root.path),
processCleanup.reason,
result,
input,
retainedOutput,
processCleanup.cleanupResult === undefined
? undefined
: {
...(processCleanup.cleanupResult.failures === undefined
? {}
: {
failures: processCleanup.cleanupResult.failures.map(
(failure) => ({ ...failure }),
),
}),
...(processCleanup.cleanupResult.unverified === undefined
? {}
: {
unverified:
processCleanup.cleanupResult.unverified.map(
(process) => ({ ...process }),
),
}),
},
),
);
}
return result;
}
const cleaned = await cleanupSqliteRoot(
@@ -360,41 +407,55 @@ const cleanupSqliteRoot = async (
return ok(null);
} catch (cause: unknown) {
return err(
new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
sqliteCleanupFailure(
[root.path],
{
reason: cause instanceof Error ? cause.message : String(cause),
previous_error: result.ok ? null : projectAnalysisError(result.error),
...(output === undefined ? {} : { captured_output: { ...output } }),
},
{
operation,
cause,
...(result.ok
? {
partialObservation: createEvidence(
result.value.subject ?? undefined,
result.value.provider,
{
operation,
parameters: sqliteInspectionParameters(input),
result: result.value.result,
confidence: "observed",
locations: result.value.locations,
limitations: result.value.limitations,
},
),
}
: result.error.partialObservation === undefined
? {}
: { partialObservation: result.error.partialObservation }),
},
cause instanceof Error ? cause.message : String(cause),
result,
input,
output,
undefined,
cause,
),
);
}
};
const sqliteCleanupFailure = (
resources: readonly string[],
reason: string,
result: SqliteOutcome,
input: InspectSqliteDatabaseInput,
output?: AnalysisCapturedOutput,
details?: Readonly<Record<string, JsonValue>>,
cause?: unknown,
): ProviderCleanupError => {
const partialObservation = result.ok
? createEvidence(result.value.subject ?? undefined, result.value.provider, {
operation,
parameters: sqliteInspectionParameters(input),
result: result.value.result,
confidence: "observed",
locations: result.value.locations,
limitations: result.value.limitations,
})
: result.error.partialObservation;
return new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
resources,
{
...details,
reason,
previous_error: result.ok ? null : projectAnalysisError(result.error),
...(output === undefined ? {} : { captured_output: { ...output } }),
},
{
operation,
...(cause === undefined ? {} : { cause }),
...(partialObservation === undefined ? {} : { partialObservation }),
},
);
};
const validateSqliteInput = (input: InspectSqliteDatabaseInput): void => {
const parsed = inspectSqliteDatabaseInputSchema.safeParse(input);
if (!parsed.success || !isAbsolute(input.path))
@@ -19,6 +19,10 @@ import { readStableArtifact } from "../artifacts/readStableArtifact.js";
import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js";
import type { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import { runOwnedCommand } from "../process/OwnedCommand.js";
import type {
ProviderProcessSupervisor,
ProviderProcessStopResult,
} from "../process/ProviderProcess.js";
import { capturedSqliteOutput } from "./SqliteDatabaseFailures.js";
import { SQLITE_DATABASE_LIMITS } from "./SqliteDatabaseLimits.js";
@@ -109,6 +113,10 @@ export const executeSqliteDatabaseWorker = async (request: {
readonly launcher?: SqliteWorkerLauncher;
readonly signal?: AbortSignal;
readonly captured: (output: AnalysisCapturedOutput) => void;
readonly retainCleanup?: (
owner: ProviderProcessSupervisor,
result: Extract<ProviderProcessStopResult, { status: "incomplete" }>,
) => void;
}): Promise<z.output<typeof inspectionSchema>> => {
const requestPath = join(request.root, "request.json");
const replyPath = join(request.root, "reply.json");
@@ -143,6 +151,9 @@ export const executeSqliteDatabaseWorker = async (request: {
{
...(request.signal === undefined ? {} : { signal: request.signal }),
...(request.launcher === undefined ? {} : { launcher: request.launcher }),
...(request.retainCleanup === undefined
? {}
: { retainCleanup: request.retainCleanup }),
},
);
const capturedOutput = capturedSqliteOutput(execution);
@@ -7,6 +7,7 @@ import { fileURLToPath } from "node:url";
import { expect, it as nativeTest, onTestFinished } from "vitest";
import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js";
import { SqliteDatabaseProvider } from "../../../src/sqlite/SqliteDatabaseProvider.js";
import { inspectSqliteDatabaseSnapshot } from "../../../src/sqlite/SqliteDatabaseInspection.js";
import { spawnOwnedProviderProcess } from "../../../src/process/ProviderProcess.js";
import { PrivateRuntimeRoot } from "../../../src/process/PrivateRuntimeRoot.js";
import { waitForProviderProcessReady } from "../../fixtures/providerProcess.js";
@@ -49,6 +50,90 @@ const createFixture = async (wal = false) => {
const expectPrivateRootRemoved = (path: string) =>
expect(access(path)).rejects.toMatchObject({ code: "ENOENT" });
it.each(["completed reply", "malformed reply"] as const)(
"preserves the primary result for a %s while retaining unconfirmed cleanup ownership",
async (replyKind) => {
const { workspace, database } = await createFixture();
let cleanupAttempts = 0;
let ownedRoot = "";
const provider = new SqliteDatabaseProvider(process.env, async (spawn) => {
ownedRoot = spawn.cwd ?? "";
const requestPath = spawn.arguments[2];
if (requestPath === undefined)
throw new Error("SQLite request path missing");
const request = JSON.parse(await readFile(requestPath, "utf8")) as {
snapshot_path: string;
reply_path: string;
input: Parameters<typeof inspectSqliteDatabaseSnapshot>[1];
};
const inspection =
replyKind === "completed reply"
? inspectSqliteDatabaseSnapshot(request.snapshot_path, request.input)
: undefined;
await writeFile(
request.reply_path,
inspection === undefined
? "{ malformed reply"
: JSON.stringify({ ok: true, inspection }),
{ flag: "wx", mode: 0o600 },
);
const launched = await spawnOwnedProviderProcess({
...spawn,
arguments: ["-e", "process.exit(0)"],
});
return {
...launched,
cleanup: async () => {
cleanupAttempts += 1;
return cleanupAttempts === 1
? { cleaned: false, reason: "fixture cleanup denial" }
: { cleaned: true, signaled: false };
},
};
});
onTestFinished(async () => {
await provider.close();
await removeTestWorkspace(workspace.root);
});
const result = await provider.inspect({
path: database.path,
table: "records",
columns: ["id"],
row_limit: 1,
});
expect(result).toMatchObject({
ok: false,
error: {
cleanupIncomplete: true,
cleanupResources: expect.arrayContaining([ownedRoot]),
diagnostics: {
reason: "fixture cleanup denial",
previous_error:
replyKind === "malformed reply" ? expect.any(Object) : null,
},
...(replyKind === "completed reply"
? {
partialObservation: {
operation: "inspect_sqlite_database",
confidence: "observed",
normalized_result: {
schema: { tables: expect.any(Array) },
rows: { returned_rows: 1 },
},
},
}
: { partialObservation: undefined }),
},
});
expect(cleanupAttempts).toBe(1);
await expect(access(`${ownedRoot}/reply.json`)).resolves.toBeUndefined();
await provider.close();
expect(cleanupAttempts).toBe(2);
await expectPrivateRootRemoved(ownedRoot);
},
);
it.each(["late-cancellation", "cleanup-failure"])(
"preserves completed native-worker observations across %s",
async (scenario) => {