Files
AgentCompany/src/main/ipc-content-operations-handlers.ts
T
ZacharyZhang-NY b1d1fbc54f Drain document lifecycle events through a durable processor
Before this commit `createDocumentWithLifecycle` enqueued ledger rows
inside the save transaction but consumed them from an in-memory
`pendingSideEffects` array straight after commit. A crash between the
commit and the drain stranded rows as `pending` forever; a crash mid
handler left them as `running`; and there was no retry budget.

The new `DocumentLifecycleProcessor` is the single drainer. It:

  - claims work via `claimPendingBatch`, an atomic `UPDATE … RETURNING`
    that flips rows from `pending` to `running` while incrementing
    `attempts`, so two drainers cannot race onto the same row;
  - dispatches each row through a handler registered by `hookType`,
    parsing the frozen `payload_json` back into the call arguments;
  - records `succeeded` / `failed` / `retryable` outcomes through the
    ledger, so a transient failure stays on the queue until the attempt
    cap expires, after which it is marked `failed` for operators;
  - reclaims rows stuck in `running` past `staleRunningMs` back to
    `pending` on every sweep tick plus at startup, so a crashed process
    never leaves a row orphaned;
  - runs a periodic sweep so every document — including those whose
    immediate post-commit drain failed — is eventually advanced.

`createDocumentWithLifecycle` now only enqueues and then calls
`lifecycleProcessor.drainForDocument` for the immediate happy-path pass.
Handlers live in `document-lifecycle-handlers.ts` so the processor stays
free of service-graph knowledge. The processor is constructed in
`bootstrap-application`, started there, and stopped on `before-quit`.
2026-04-19 16:14:51 -04:00

215 lines
8.0 KiB
TypeScript

import {
IPC_CHANNELS,
deleteEntitySchema,
documentInputSchema,
knowledgeEntryInputSchema,
meetingInputSchema,
sprintInputSchema,
} from "@shared/contracts";
import { fail, ok, type RegisterDomainHandlersDependencies } from "./ipc-domain-common";
import { createDocumentWithLifecycle } from "./document-lifecycle";
const COMPANY_SCOPED_REFERENCE_TABLES = new Set([
"agents",
"goals",
"projects",
"tasks",
"approvals",
"runs",
"secrets",
"meetings",
"documents",
"knowledge_base",
"sprints",
"automation_rules",
"workflow_pipelines",
]);
export function registerContentOperationsHandlers({
registerHandle,
db,
ensureCompanyExists,
findInvalidReference,
publishDomainChanged,
dispatchAutomation,
threadIntegration,
checkpointEngine,
lifecycleProcessor,
logger,
}: Pick<
RegisterDomainHandlersDependencies,
| "registerHandle"
| "db"
| "ensureCompanyExists"
| "findInvalidReference"
| "publishDomainChanged"
| "dispatchAutomation"
| "threadIntegration"
| "checkpointEngine"
| "lifecycleProcessor"
| "logger"
>) {
registerHandle(IPC_CHANNELS.saveMeeting, async (_event, payload) => {
const parsed = meetingInputSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
let participantAgentIds: string[] = [];
try {
const parsedParticipantIds = JSON.parse(parsed.participantAgentIds);
participantAgentIds = Array.isArray(parsedParticipantIds)
? parsedParticipantIds.filter((value): value is string => typeof value === "string")
: [];
} catch (err) {
logger.warn(`[ipc/content] Meeting participants JSON parse failed:`, err instanceof Error ? err.message : String(err));
return fail("INVALID_PARTICIPANTS", "Meeting participants must be valid JSON.");
}
const invalidReference = findInvalidReference(parsed.companyId, [
{ id: parsed.organizerAgentId, table: "agents", label: "meeting organizer" },
...participantAgentIds.map((agentId) => ({ id: agentId, table: "agents", label: "meeting participant" })),
]);
if (invalidReference) {
return fail("INVALID_COMPANY_REFERENCE", `The selected ${invalidReference.label} does not belong to the current company.`);
}
const id = db.saveMeeting(parsed);
publishDomainChanged();
return ok(id);
});
registerHandle(IPC_CHANNELS.deleteMeeting, async (_event, payload) => {
const parsed = deleteEntitySchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
if (!db.belongsToCompany("meetings", parsed.id, parsed.companyId)) {
return fail("ENTITY_NOT_FOUND", "Meeting not found in the current company.");
}
db.deleteMeeting(parsed.id, parsed.companyId);
publishDomainChanged();
return ok(true);
});
registerHandle(IPC_CHANNELS.saveDocument, async (_event, payload) => {
const parsed = documentInputSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const invalidReference = findInvalidReference(parsed.companyId, [
{ id: parsed.authorAgentId, table: "agents", label: "document author" },
{ id: parsed.reviewerAgentId, table: "agents", label: "document reviewer" },
{ id: parsed.projectId, table: "projects", label: "document project" },
{ id: parsed.goalId, table: "goals", label: "document goal" },
{ id: parsed.parentDocId, table: "documents", label: "parent document" },
]);
if (invalidReference) {
return fail("INVALID_COMPANY_REFERENCE", `The selected ${invalidReference.label} does not belong to the current company.`);
}
const outcome = createDocumentWithLifecycle(
{
db,
threadIntegration,
checkpointEngine,
lifecycleProcessor,
publishDomainChanged,
logger,
},
{
id: parsed.id ?? null,
companyId: parsed.companyId,
type: parsed.type,
title: parsed.title,
content: parsed.content ?? "",
authorAgentId: parsed.authorAgentId ?? null,
reviewerAgentId: parsed.reviewerAgentId ?? null,
projectId: parsed.projectId ?? null,
goalId: parsed.goalId ?? null,
parentDocId: parsed.parentDocId ?? null,
version: parsed.version ?? 1,
status: parsed.status ?? "draft",
tagsJson: parsed.tagsJson ?? "[]",
sourceRunId: null,
},
);
return ok(outcome.id);
});
registerHandle(IPC_CHANNELS.deleteDocument, async (_event, payload) => {
const parsed = deleteEntitySchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
if (!db.belongsToCompany("documents", parsed.id, parsed.companyId)) {
return fail("ENTITY_NOT_FOUND", "Document not found in the current company.");
}
db.deleteDocument(parsed.id, parsed.companyId);
publishDomainChanged();
return ok(true);
});
registerHandle(IPC_CHANNELS.saveKnowledgeEntry, async (_event, payload) => {
const parsed = knowledgeEntryInputSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const knowledgeReferences = [
{ id: parsed.authorAgentId, table: "agents", label: "knowledge author" },
...(parsed.referencedEntityId && parsed.referencedEntityType
&& COMPANY_SCOPED_REFERENCE_TABLES.has(parsed.referencedEntityType)
? [{ id: parsed.referencedEntityId, table: parsed.referencedEntityType, label: "referenced entity" }]
: []),
];
const invalidReference = findInvalidReference(parsed.companyId, [
...knowledgeReferences,
]);
if (invalidReference) {
return fail("INVALID_COMPANY_REFERENCE", `The selected ${invalidReference.label} does not belong to the current company.`);
}
const id = db.saveKnowledgeEntry(parsed);
publishDomainChanged();
return ok(id);
});
registerHandle(IPC_CHANNELS.deleteKnowledgeEntry, async (_event, payload) => {
const parsed = deleteEntitySchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
if (!db.belongsToCompany("knowledge_base", parsed.id, parsed.companyId)) {
return fail("ENTITY_NOT_FOUND", "Knowledge entry not found in the current company.");
}
db.deleteKnowledgeEntry(parsed.id, parsed.companyId);
publishDomainChanged();
return ok(true);
});
registerHandle(IPC_CHANNELS.saveSprint, async (_event, payload) => {
const parsed = sprintInputSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
let previousSprintStatus: string | null = null;
if (parsed.id) {
try {
previousSprintStatus = db.listSnapshot().sprints.find((s: { id: string }) => s.id === parsed.id)?.status ?? null;
} catch (err) {
logger.warn(`[ipc/content] listSnapshot failed while resolving sprint ${parsed.id} previous status:`, err instanceof Error ? err.message : String(err));
previousSprintStatus = null;
}
}
const id = db.saveSprint(parsed);
if (parsed.status === "active" && previousSprintStatus !== "active") {
dispatchAutomation("sprint_started", { companyId: parsed.companyId });
}
if (parsed.status === "completed" && previousSprintStatus !== "completed") {
dispatchAutomation("sprint_ended", { companyId: parsed.companyId });
}
publishDomainChanged();
return ok(id);
});
registerHandle(IPC_CHANNELS.deleteSprint, async (_event, payload) => {
const parsed = deleteEntitySchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
if (!db.belongsToCompany("sprints", parsed.id, parsed.companyId)) {
return fail("ENTITY_NOT_FOUND", "Sprint not found in the current company.");
}
db.deleteSprint(parsed.id, parsed.companyId);
publishDomainChanged();
return ok(true);
});
}