fix(memory-jobs): address review and CI feedback

This commit is contained in:
Dotta committed 2026-04-21 15:19:58 -05:00
1 parent 40dd54c96a
commit 6dd968d030
10 files changed
+118 -63

No files matched your search

@@ -32,36 +32,6 @@ CREATE TABLE IF NOT EXISTS "memory_extraction_jobs" (
"updated_at" timestamp with time zone DEFAULT now() NOT NULL
);
--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "company_id" uuid NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "binding_id" uuid NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "binding_key" text NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "operation_type" text NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "status" text DEFAULT 'queued' NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_agent_id" uuid;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_issue_id" uuid;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_project_id" uuid;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_goal_id" uuid;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_heartbeat_run_id" uuid;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "hook_kind" text;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "provider_job_id" text;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "submitted_at" timestamp with time zone DEFAULT now() NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "started_at" timestamp with time zone;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "finished_at" timestamp with time zone;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "attribution_mode" text DEFAULT 'untracked' NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "cost_cents" integer DEFAULT 0 NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "result_summary" text;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "error_code" text;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "error" text;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_kind" text NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_ref_json" jsonb;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "retry_of_job_id" uuid;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "attempt_number" integer DEFAULT 1 NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "dispatcher_kind" text DEFAULT 'in_process' NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "lease_expires_at" timestamp with time zone;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "usage_json" jsonb;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "result_json" jsonb;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "created_at" timestamp with time zone DEFAULT now() NOT NULL;--> statement-breakpoint
ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "updated_at" timestamp with time zone DEFAULT now() NOT NULL;--> statement-breakpoint
DROP INDEX IF EXISTS "issues_open_routine_execution_uq";--> statement-breakpoint
ALTER TABLE "issues" ADD COLUMN IF NOT EXISTS "origin_fingerprint" text DEFAULT 'default' NOT NULL;--> statement-breakpoint
ALTER TABLE "routine_runs" ADD COLUMN IF NOT EXISTS "dispatch_fingerprint" text;--> statement-breakpoint
@@ -71,7 +41,7 @@ DO $$ BEGIN
WHERE conname = 'memory_extraction_jobs_company_id_companies_id_fk'
AND conrelid = 'memory_extraction_jobs'::regclass
) THEN
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE no action ON UPDATE no action;
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
@@ -80,7 +50,7 @@ DO $$ BEGIN
WHERE conname = 'memory_extraction_jobs_source_agent_id_agents_id_fk'
AND conrelid = 'memory_extraction_jobs'::regclass
) THEN
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_agent_id_agents_id_fk" FOREIGN KEY ("source_agent_id") REFERENCES "public"."agents"("id") ON DELETE no action ON UPDATE no action;
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_agent_id_agents_id_fk" FOREIGN KEY ("source_agent_id") REFERENCES "public"."agents"("id") ON DELETE set null ON UPDATE no action;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
@@ -89,7 +59,7 @@ DO $$ BEGIN
WHERE conname = 'memory_extraction_jobs_source_issue_id_issues_id_fk'
AND conrelid = 'memory_extraction_jobs'::regclass
) THEN
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_issue_id_issues_id_fk" FOREIGN KEY ("source_issue_id") REFERENCES "public"."issues"("id") ON DELETE no action ON UPDATE no action;
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_issue_id_issues_id_fk" FOREIGN KEY ("source_issue_id") REFERENCES "public"."issues"("id") ON DELETE set null ON UPDATE no action;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
@@ -98,7 +68,7 @@ DO $$ BEGIN
WHERE conname = 'memory_extraction_jobs_source_project_id_projects_id_fk'
AND conrelid = 'memory_extraction_jobs'::regclass
) THEN
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_project_id_projects_id_fk" FOREIGN KEY ("source_project_id") REFERENCES "public"."projects"("id") ON DELETE no action ON UPDATE no action;
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_project_id_projects_id_fk" FOREIGN KEY ("source_project_id") REFERENCES "public"."projects"("id") ON DELETE set null ON UPDATE no action;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
@@ -107,7 +77,7 @@ DO $$ BEGIN
WHERE conname = 'memory_extraction_jobs_source_goal_id_goals_id_fk'
AND conrelid = 'memory_extraction_jobs'::regclass
) THEN
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_goal_id_goals_id_fk" FOREIGN KEY ("source_goal_id") REFERENCES "public"."goals"("id") ON DELETE no action ON UPDATE no action;
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_goal_id_goals_id_fk" FOREIGN KEY ("source_goal_id") REFERENCES "public"."goals"("id") ON DELETE set null ON UPDATE no action;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
@@ -116,7 +86,7 @@ DO $$ BEGIN
WHERE conname = 'memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk'
AND conrelid = 'memory_extraction_jobs'::regclass
) THEN
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("source_heartbeat_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE no action ON UPDATE no action;
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("source_heartbeat_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE set null ON UPDATE no action;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
@@ -125,7 +95,7 @@ DO $$ BEGIN
WHERE conname = 'memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk'
AND conrelid = 'memory_extraction_jobs'::regclass
) THEN
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk" FOREIGN KEY ("retry_of_job_id") REFERENCES "public"."memory_extraction_jobs"("id") ON DELETE no action ON UPDATE no action;
ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk" FOREIGN KEY ("retry_of_job_id") REFERENCES "public"."memory_extraction_jobs"("id") ON DELETE set null ON UPDATE no action;
END IF;
END $$;--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_company_status_submitted_idx" ON "memory_extraction_jobs" USING btree ("company_id","status","submitted_at");--> statement-breakpoint
@@ -10704,7 +10704,7 @@
"columnsTo": [
"id"
],
"onDelete": "no action",
"onDelete": "cascade",
"onUpdate": "no action"
},
"memory_extraction_jobs_source_agent_id_agents_id_fk": {
@@ -10717,7 +10717,7 @@
"columnsTo": [
"id"
],
"onDelete": "no action",
"onDelete": "set null",
"onUpdate": "no action"
},
"memory_extraction_jobs_source_issue_id_issues_id_fk": {
@@ -10730,7 +10730,7 @@
"columnsTo": [
"id"
],
"onDelete": "no action",
"onDelete": "set null",
"onUpdate": "no action"
},
"memory_extraction_jobs_source_project_id_projects_id_fk": {
@@ -10743,7 +10743,7 @@
"columnsTo": [
"id"
],
"onDelete": "no action",
"onDelete": "set null",
"onUpdate": "no action"
},
"memory_extraction_jobs_source_goal_id_goals_id_fk": {
@@ -10756,7 +10756,7 @@
"columnsTo": [
"id"
],
"onDelete": "no action",
"onDelete": "set null",
"onUpdate": "no action"
},
"memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk": {
@@ -10769,7 +10769,7 @@
"columnsTo": [
"id"
],
"onDelete": "no action",
"onDelete": "set null",
"onUpdate": "no action"
},
"memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk": {
@@ -27,16 +27,16 @@ export const memoryExtractionJobs = pgTable(
"memory_extraction_jobs",
{
id: uuid("id").primaryKey().defaultRandom(),
companyId: uuid("company_id").notNull().references(() => companies.id),
companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }),
bindingId: uuid("binding_id").notNull(),
bindingKey: text("binding_key").notNull(),
operationType: text("operation_type").$type<MemoryExtractionJobOperationType>().notNull(),
status: text("status").$type<MemoryExtractionJobStatus>().notNull().default("queued"),
sourceAgentId: uuid("source_agent_id").references(() => agents.id),
sourceIssueId: uuid("source_issue_id").references(() => issues.id),
sourceProjectId: uuid("source_project_id").references(() => projects.id),
sourceGoalId: uuid("source_goal_id").references(() => goals.id),
sourceHeartbeatRunId: uuid("source_heartbeat_run_id").references(() => heartbeatRuns.id),
sourceAgentId: uuid("source_agent_id").references(() => agents.id, { onDelete: "set null" }),
sourceIssueId: uuid("source_issue_id").references(() => issues.id, { onDelete: "set null" }),
sourceProjectId: uuid("source_project_id").references(() => projects.id, { onDelete: "set null" }),
sourceGoalId: uuid("source_goal_id").references(() => goals.id, { onDelete: "set null" }),
sourceHeartbeatRunId: uuid("source_heartbeat_run_id").references(() => heartbeatRuns.id, { onDelete: "set null" }),
hookKind: text("hook_kind").$type<MemoryExtractionJobHookKind>(),
providerJobId: text("provider_job_id"),
submittedAt: timestamp("submitted_at", { withTimezone: true }).notNull().defaultNow(),
+7
View File
@@ -40,6 +40,13 @@ export const listMemoryExtractionJobsQuerySchema = z
})
.strict()
.superRefine((value, ctx) => {
if (value.status && value.effectiveState) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
path: ["effectiveState"],
message: "effectiveState and status cannot be used together",
});
}
if (value.submittedAfter && value.submittedBefore && value.submittedAfter > value.submittedBefore) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
@@ -35,6 +35,22 @@ async function insertCompany(db: ReturnType<typeof createDb>, companyId: string)
});
}
async function waitForJobStatus(
store: ReturnType<typeof memoryJobStore>,
companyId: string,
jobId: string,
status: string,
) {
for (let attempt = 0; attempt < 20; attempt += 1) {
const detail = await store.getDetail(companyId, jobId);
if (detail?.status === status) {
return detail;
}
await new Promise((resolve) => setTimeout(resolve, 10));
}
return store.getDetail(companyId, jobId);
}
describeEmbeddedPostgres("memory job capture", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startMemoryJobTestDatabase>> | null = null;
@@ -193,7 +209,7 @@ describeEmbeddedPostgres("memory job capture", () => {
await dispatcher.tick();
const completed = await store.getDetail(companyId, queued.id);
const completed = await waitForJobStatus(store, companyId, queued.id, "succeeded");
expect(completed).toMatchObject({
id: queued.id,
status: "succeeded",
@@ -30,6 +30,23 @@ async function insertCompany(db: ReturnType<typeof createDb>, companyId: string)
});
}
async function waitForJobStatus(
store: ReturnType<typeof memoryJobStore>,
companyId: string,
jobId: string,
status: string,
now: Date,
) {
for (let attempt = 0; attempt < 20; attempt += 1) {
const detail = await store.getDetail(companyId, jobId, { now });
if (detail?.status === status) {
return detail;
}
await new Promise((resolve) => setTimeout(resolve, 10));
}
return store.getDetail(companyId, jobId, { now });
}
describeEmbeddedPostgres("memory job dispatcher", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startMemoryJobTestDatabase>> | null = null;
@@ -83,7 +100,7 @@ describeEmbeddedPostgres("memory job dispatcher", () => {
await dispatcher.tick();
const detail = await store.getDetail(companyId, queued.id, { now: currentTime });
const detail = await waitForJobStatus(store, companyId, queued.id, "succeeded", currentTime);
expect(detail).toMatchObject({
id: queued.id,
@@ -132,7 +149,7 @@ describeEmbeddedPostgres("memory job dispatcher", () => {
await dispatcher.tick();
const runningDetail = await store.getDetail(companyId, queued.id, { now: currentTime });
const runningDetail = await waitForJobStatus(store, companyId, queued.id, "running", currentTime);
expect(runningDetail).toMatchObject({
id: queued.id,
status: "running",
@@ -140,6 +140,25 @@ describeEmbeddedPostgres("memory job routes", () => {
});
});
it("rejects ambiguous status and effective-state filters", async () => {
const companyId = randomUUID();
await insertCompany(db, companyId);
const app = createApp(db, {
type: "board",
userId: "board",
source: "local_implicit",
isInstanceAdmin: true,
});
const res = await request(app).get(
`/api/companies/${companyId}/memory/jobs?status=succeeded&effectiveState=stuck`,
);
expect(res.status).toBe(400);
expect(res.body.error).toContain("effectiveState and status cannot be used together");
});
it("returns job detail scoped to the requested company", async () => {
const companyId = randomUUID();
const jobId = randomUUID();
@@ -144,6 +144,32 @@ vi.mock("../services/index.js", () => ({
})),
}));
vi.mock("../services/memory-job-capture.js", () => ({
createPostRunMemoryCaptureHandler: vi.fn(() => vi.fn()),
}));
vi.mock("../services/memory-job-dispatcher.js", () => ({
createMemoryJobDispatcher: vi.fn(() => ({
start: vi.fn(),
stop: vi.fn(),
tick: vi.fn(async () => undefined),
sweepExpiredLeases: vi.fn(async () => ({ recovered: 0, jobIds: [] })),
diagnostics: vi.fn(() => ({
running: false,
activeJobCount: 0,
activeJobIds: [],
tickCount: 0,
lastTickAt: null,
recoveryCount: 0,
lastRecoveryAt: null,
})),
})),
}));
vi.mock("../services/memory-job-store.js", () => ({
memoryJobStore: vi.fn(() => ({})),
}));
vi.mock("../storage/index.js", () => ({
createStorageServiceFromConfig: vi.fn(() => ({ id: "storage-service" })),
}));
+3 -1
View File
@@ -69,7 +69,9 @@ function readFiniteField(record: Record<string, unknown> | null, key: string): n
}
function formatUuidFromHex(hex: string): string {
return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20, 32)}`;
const version = `4${hex.slice(13, 16)}`;
const variant = `${((parseInt(hex[16] ?? "0", 16) & 0x3) | 0x8).toString(16)}${hex.slice(17, 20)}`;
return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${version}-${variant}-${hex.slice(20, 32)}`;
}
export function buildSyntheticMemoryBindingId(companyId: string, bindingKey: string): string {
+8 -10
View File
@@ -58,6 +58,10 @@ export interface MemoryJobDispatcher {
diagnostics(): MemoryJobDispatcherDiagnostics;
}
function childLogger(bindings: Record<string, unknown>): typeof logger {
return typeof logger.child === "function" ? logger.child(bindings) : logger;
}
export function createMemoryJobDispatcher(
options: MemoryJobDispatcherOptions,
): MemoryJobDispatcher {
@@ -66,7 +70,7 @@ export function createMemoryJobDispatcher(
const leaseDurationMs = options.leaseDurationMs ?? DEFAULT_LEASE_DURATION_MS;
const maxConcurrentJobs = options.maxConcurrentJobs ?? DEFAULT_MAX_CONCURRENT_JOBS;
const now = options.now ?? (() => new Date());
const log = logger.child({ service: "memory_job_dispatcher" });
const log = childLogger({ service: "memory_job_dispatcher" });
let pollTimer: ReturnType<typeof setInterval> | null = null;
let recoveryTimer: ReturnType<typeof setInterval> | null = null;
@@ -81,12 +85,12 @@ export function createMemoryJobDispatcher(
const activeJobs = new Set<string>();
async function runClaimedJob(job: ClaimedMemoryJob): Promise<void> {
const jobLog = log.child({
const jobLog = typeof log.child === "function" ? log.child({
jobId: job.id,
companyId: job.companyId,
bindingKey: job.bindingKey,
operationType: job.operationType,
});
}) : log;
try {
const handler = await options.resolveHandler(job);
@@ -185,8 +189,6 @@ export function createMemoryJobDispatcher(
tickCount += 1;
lastTickAt = now();
const dispatches: Promise<void>[] = [];
try {
while (activeJobs.size < maxConcurrentJobs) {
const claimed = await options.store.claimNext({
@@ -200,11 +202,7 @@ export function createMemoryJobDispatcher(
}
activeJobs.add(claimed.id);
dispatches.push(runClaimedJob(claimed));
}
if (dispatches.length > 0) {
await Promise.allSettled(dispatches);
void runClaimedJob(claimed);
}
} catch (error) {
log.error(