feat(memory-jobs): add async post-run memory capture job state

Introduces the `memory_extraction_jobs` table plus its schema, store,
in-process dispatcher with recovery sweep, HTTP routes, and UI surface,
and wires the heartbeat run-completion path to enqueue a post-run
capture job with full provenance (agent/issue/project/goal/run), binding
key, usage summary, result summary, and direct-vs-included cost
attribution. A single post-run capture handler is registered on server
startup; rerun behavior for capture-originated jobs is covered by the
store/dispatcher.

Closes PAP-1705, PAP-1706, PAP-1707.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-04-21 14:55:50 -05:00
1 parent cb94164e19
commit 2c8777a319
30 files changed
+18473 -2

No files matched your search

@@ -0,0 +1,143 @@
CREATE TABLE IF NOT EXISTS "memory_extraction_jobs" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"company_id" uuid NOT NULL,
"binding_id" uuid NOT NULL,
"binding_key" text NOT NULL,
"operation_type" text NOT NULL,
"status" text DEFAULT 'queued' NOT NULL,
"source_agent_id" uuid,
"source_issue_id" uuid,
"source_project_id" uuid,
"source_goal_id" uuid,
"source_heartbeat_run_id" uuid,
"hook_kind" text,
"provider_job_id" text,
"submitted_at" timestamp with time zone DEFAULT now() NOT NULL,
"started_at" timestamp with time zone,
"finished_at" timestamp with time zone,
"attribution_mode" text DEFAULT 'untracked' NOT NULL,
"cost_cents" integer DEFAULT 0 NOT NULL,
"result_summary" text,
"error_code" text,
"error" text,
"source_kind" text NOT NULL,
"source_ref_json" jsonb,
"retry_of_job_id" uuid,
"attempt_number" integer DEFAULT 1 NOT NULL,
"dispatcher_kind" text DEFAULT 'in_process' NOT NULL,
"lease_expires_at" timestamp with time zone,
"usage_json" jsonb,
"result_json" jsonb,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"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
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
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;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
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;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
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;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
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;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
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;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
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;
END IF;
END $$;--> statement-breakpoint
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
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;
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
CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_company_binding_submitted_idx" ON "memory_extraction_jobs" USING btree ("company_id","binding_key","submitted_at");--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_company_operation_submitted_idx" ON "memory_extraction_jobs" USING btree ("company_id","operation_type","submitted_at");--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_company_issue_submitted_idx" ON "memory_extraction_jobs" USING btree ("company_id","source_issue_id","submitted_at");--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_company_run_submitted_idx" ON "memory_extraction_jobs" USING btree ("company_id","source_heartbeat_run_id","submitted_at");--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_company_status_lease_expires_idx" ON "memory_extraction_jobs" USING btree ("company_id","status","lease_expires_at");--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_retry_of_job_idx" ON "memory_extraction_jobs" USING btree ("retry_of_job_id");--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "routine_runs_dispatch_fingerprint_idx" ON "routine_runs" USING btree ("routine_id","dispatch_fingerprint");--> statement-breakpoint
CREATE UNIQUE INDEX IF NOT EXISTS "issues_open_routine_execution_uq" ON "issues" USING btree ("company_id","origin_kind","origin_id","origin_fingerprint") WHERE "issues"."origin_kind" = 'routine_execution'
and "issues"."origin_id" is not null
and "issues"."hidden_at" is null
and "issues"."execution_run_id" is not null
and "issues"."status" in ('backlog', 'todo', 'in_progress', 'in_review', 'blocked');
File diff suppressed because it is too large. Load diff
@@ -442,6 +442,13 @@
"when": 1776780000000,
"tag": "0062_routine_run_dispatch_fingerprint",
"breakpoints": true
},
{
"idx": 63,
"version": "7",
"when": 1776793732272,
"tag": "0063_motionless_gertrude_yorkes",
"breakpoints": true
}
]
}
}
+1
View File
@@ -46,6 +46,7 @@ export { issueAttachments } from "./issue_attachments.js";
export { documents } from "./documents.js";
export { documentRevisions } from "./document_revisions.js";
export { issueDocuments } from "./issue_documents.js";
export { memoryExtractionJobs } from "./memory_extraction_jobs.js";
export { heartbeatRuns } from "./heartbeat_runs.js";
export { heartbeatRunEvents } from "./heartbeat_run_events.js";
export { costEvents } from "./cost_events.js";
@@ -0,0 +1,100 @@
import {
type AnyPgColumn,
index,
integer,
jsonb,
pgTable,
text,
timestamp,
uuid,
} from "drizzle-orm/pg-core";
import type {
MemoryExtractionJobAttributionMode,
MemoryExtractionJobDispatcherKind,
MemoryExtractionJobHookKind,
MemoryExtractionJobOperationType,
MemoryExtractionJobSourceKind,
MemoryExtractionJobStatus,
} from "@paperclipai/shared";
import { agents } from "./agents.js";
import { companies } from "./companies.js";
import { goals } from "./goals.js";
import { heartbeatRuns } from "./heartbeat_runs.js";
import { issues } from "./issues.js";
import { projects } from "./projects.js";
export const memoryExtractionJobs = pgTable(
"memory_extraction_jobs",
{
id: uuid("id").primaryKey().defaultRandom(),
companyId: uuid("company_id").notNull().references(() => companies.id),
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),
hookKind: text("hook_kind").$type<MemoryExtractionJobHookKind>(),
providerJobId: text("provider_job_id"),
submittedAt: timestamp("submitted_at", { withTimezone: true }).notNull().defaultNow(),
startedAt: timestamp("started_at", { withTimezone: true }),
finishedAt: timestamp("finished_at", { withTimezone: true }),
attributionMode: text("attribution_mode")
.$type<MemoryExtractionJobAttributionMode>()
.notNull()
.default("untracked"),
costCents: integer("cost_cents").notNull().default(0),
resultSummary: text("result_summary"),
errorCode: text("error_code"),
error: text("error"),
sourceKind: text("source_kind").$type<MemoryExtractionJobSourceKind>().notNull(),
sourceRefJson: jsonb("source_ref_json").$type<Record<string, unknown>>(),
retryOfJobId: uuid("retry_of_job_id").references((): AnyPgColumn => memoryExtractionJobs.id),
attemptNumber: integer("attempt_number").notNull().default(1),
dispatcherKind: text("dispatcher_kind")
.$type<MemoryExtractionJobDispatcherKind>()
.notNull()
.default("in_process"),
leaseExpiresAt: timestamp("lease_expires_at", { withTimezone: true }),
usageJson: jsonb("usage_json").$type<Record<string, unknown>>(),
resultJson: jsonb("result_json").$type<Record<string, unknown>>(),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
},
(table) => ({
companyStatusSubmittedIdx: index("memory_extraction_jobs_company_status_submitted_idx").on(
table.companyId,
table.status,
table.submittedAt,
),
companyBindingSubmittedIdx: index("memory_extraction_jobs_company_binding_submitted_idx").on(
table.companyId,
table.bindingKey,
table.submittedAt,
),
companyOperationSubmittedIdx: index("memory_extraction_jobs_company_operation_submitted_idx").on(
table.companyId,
table.operationType,
table.submittedAt,
),
companyIssueSubmittedIdx: index("memory_extraction_jobs_company_issue_submitted_idx").on(
table.companyId,
table.sourceIssueId,
table.submittedAt,
),
companyRunSubmittedIdx: index("memory_extraction_jobs_company_run_submitted_idx").on(
table.companyId,
table.sourceHeartbeatRunId,
table.submittedAt,
),
companyStatusLeaseExpiresIdx: index("memory_extraction_jobs_company_status_lease_expires_idx").on(
table.companyId,
table.status,
table.leaseExpiresAt,
),
retryOfJobIdx: index("memory_extraction_jobs_retry_of_job_idx").on(table.retryOfJobId),
}),
);
+52
View File
@@ -326,6 +326,58 @@ export const BUDGET_INCIDENT_RESOLUTION_ACTIONS = [
] as const;
export type BudgetIncidentResolutionAction = (typeof BUDGET_INCIDENT_RESOLUTION_ACTIONS)[number];
export const MEMORY_EXTRACTION_JOB_OPERATION_TYPES = [
"capture",
"compaction",
"dreaming",
] as const;
export type MemoryExtractionJobOperationType = (typeof MEMORY_EXTRACTION_JOB_OPERATION_TYPES)[number];
export const MEMORY_EXTRACTION_JOB_STATUSES = [
"queued",
"running",
"succeeded",
"failed",
"cancelled",
] as const;
export type MemoryExtractionJobStatus = (typeof MEMORY_EXTRACTION_JOB_STATUSES)[number];
export const MEMORY_EXTRACTION_JOB_HOOK_KINDS = [
"post_run_capture",
"issue_comment_capture",
"issue_document_capture",
] as const;
export type MemoryExtractionJobHookKind = (typeof MEMORY_EXTRACTION_JOB_HOOK_KINDS)[number];
export const MEMORY_EXTRACTION_JOB_SOURCE_KINDS = [
"run",
"comment",
"document",
"manual",
"maintenance",
] as const;
export type MemoryExtractionJobSourceKind = (typeof MEMORY_EXTRACTION_JOB_SOURCE_KINDS)[number];
export const MEMORY_EXTRACTION_JOB_DISPATCHER_KINDS = [
"in_process",
"host_worker",
] as const;
export type MemoryExtractionJobDispatcherKind = (typeof MEMORY_EXTRACTION_JOB_DISPATCHER_KINDS)[number];
export const MEMORY_EXTRACTION_JOB_ATTRIBUTION_MODES = [
"included_in_run",
"billed_directly",
"external_invoice",
"untracked",
] as const;
export type MemoryExtractionJobAttributionMode = (typeof MEMORY_EXTRACTION_JOB_ATTRIBUTION_MODES)[number];
export const MEMORY_EXTRACTION_JOB_EFFECTIVE_STATES = [
...MEMORY_EXTRACTION_JOB_STATUSES,
"stuck",
] as const;
export type MemoryExtractionJobEffectiveState = (typeof MEMORY_EXTRACTION_JOB_EFFECTIVE_STATES)[number];
export const HEARTBEAT_INVOCATION_SOURCES = [
"timer",
"assignment",
+42
View File
@@ -182,6 +182,23 @@ export {
type PluginBridgeErrorCode,
} from "./constants.js";
export {
MEMORY_EXTRACTION_JOB_ATTRIBUTION_MODES,
MEMORY_EXTRACTION_JOB_DISPATCHER_KINDS,
MEMORY_EXTRACTION_JOB_EFFECTIVE_STATES,
MEMORY_EXTRACTION_JOB_HOOK_KINDS,
MEMORY_EXTRACTION_JOB_OPERATION_TYPES,
MEMORY_EXTRACTION_JOB_SOURCE_KINDS,
MEMORY_EXTRACTION_JOB_STATUSES,
type MemoryExtractionJobAttributionMode,
type MemoryExtractionJobDispatcherKind,
type MemoryExtractionJobEffectiveState,
type MemoryExtractionJobHookKind,
type MemoryExtractionJobOperationType,
type MemoryExtractionJobSourceKind,
type MemoryExtractionJobStatus,
} from "./constants.js";
export {
ALL_INTERFACES_BIND_HOST,
LOOPBACK_BIND_HOST,
@@ -437,6 +454,17 @@ export type {
QuotaWindow,
ProviderQuotaResult,
} from "./types/index.js";
export type {
MemoryExtractionJob,
MemoryExtractionJobDetail,
MemoryExtractionJobDetailResponse,
MemoryExtractionJobListItem,
MemoryExtractionJobListResponse,
MemoryExtractionJobRerunResponse,
MemoryExtractionJobSourceRef,
MemoryExtractionJobUsage,
} from "./types/memory.js";
export {
ISSUE_REFERENCE_IDENTIFIER_RE,
buildIssueReferenceHref,
@@ -746,6 +774,20 @@ export {
type ListPluginState,
} from "./validators/index.js";
export {
listMemoryExtractionJobsQuerySchema,
memoryExtractionJobAttributionModeSchema,
memoryExtractionJobDispatcherKindSchema,
memoryExtractionJobEffectiveStateSchema,
memoryExtractionJobHookKindSchema,
memoryExtractionJobOperationTypeSchema,
memoryExtractionJobSourceKindSchema,
memoryExtractionJobStatusSchema,
rerunMemoryExtractionJobSchema,
type ListMemoryExtractionJobsQuery,
type RerunMemoryExtractionJob,
} from "./validators/memory.js";
export { API_PREFIX, API } from "./api.js";
export { normalizeAgentUrlKey, deriveAgentUrlKey, isUuidLike } from "./agent-url-key.js";
export { deriveProjectUrlKey, normalizeProjectUrlKey, hasNonAsciiContent } from "./project-url-key.js";
+10
View File
@@ -159,6 +159,16 @@ export type {
} from "./routine.js";
export type { CostEvent, CostSummary, CostByAgent, CostByProviderModel, CostByBiller, CostByAgentModel, CostWindowSpendRow, CostByProject } from "./cost.js";
export type { FinanceEvent, FinanceSummary, FinanceByBiller, FinanceByKind } from "./finance.js";
export type {
MemoryExtractionJob,
MemoryExtractionJobDetail,
MemoryExtractionJobDetailResponse,
MemoryExtractionJobListItem,
MemoryExtractionJobListResponse,
MemoryExtractionJobRerunResponse,
MemoryExtractionJobSourceRef,
MemoryExtractionJobUsage,
} from "./memory.js";
export type {
AgentWakeupResponse,
AgentWakeupSkipped,
+112
View File
@@ -0,0 +1,112 @@
import type {
BillingType,
MemoryExtractionJobAttributionMode,
MemoryExtractionJobDispatcherKind,
MemoryExtractionJobEffectiveState,
MemoryExtractionJobHookKind,
MemoryExtractionJobOperationType,
MemoryExtractionJobSourceKind,
MemoryExtractionJobStatus,
} from "../constants.js";
export interface MemoryExtractionJobSourceRef {
commentId?: string | null;
documentKey?: string | null;
subjectId?: string | null;
namespace?: string | null;
[key: string]: unknown;
}
export interface MemoryExtractionJobUsage {
provider: string;
biller?: string | null;
model?: string | null;
billingType?: BillingType | null;
inputTokens?: number | null;
cachedInputTokens?: number | null;
outputTokens?: number | null;
embeddingTokens?: number | null;
costCents?: number | null;
latencyMs?: number | null;
details?: Record<string, unknown> | null;
}
export interface MemoryExtractionJob {
id: string;
companyId: string;
bindingId: string;
bindingKey: string;
operationType: MemoryExtractionJobOperationType;
status: MemoryExtractionJobStatus;
sourceAgentId: string | null;
sourceIssueId: string | null;
sourceProjectId: string | null;
sourceGoalId: string | null;
sourceHeartbeatRunId: string | null;
hookKind: MemoryExtractionJobHookKind | null;
providerJobId: string | null;
submittedAt: Date;
startedAt: Date | null;
finishedAt: Date | null;
attributionMode: MemoryExtractionJobAttributionMode;
costCents: number;
resultSummary: string | null;
errorCode: string | null;
error: string | null;
sourceKind: MemoryExtractionJobSourceKind;
sourceRefJson: MemoryExtractionJobSourceRef | null;
retryOfJobId: string | null;
attemptNumber: number;
dispatcherKind: MemoryExtractionJobDispatcherKind;
leaseExpiresAt: Date | null;
usageJson: MemoryExtractionJobUsage | null;
resultJson: Record<string, unknown> | null;
createdAt: Date;
updatedAt: Date;
}
export interface MemoryExtractionJobListItem {
id: string;
bindingId: string;
bindingKey: string;
operationType: MemoryExtractionJobOperationType;
status: MemoryExtractionJobStatus;
effectiveState: MemoryExtractionJobEffectiveState;
sourceAgentId: string | null;
sourceIssueId: string | null;
sourceProjectId: string | null;
sourceGoalId: string | null;
sourceHeartbeatRunId: string | null;
hookKind: MemoryExtractionJobHookKind | null;
providerJobId: string | null;
submittedAt: Date;
startedAt: Date | null;
finishedAt: Date | null;
attributionMode: MemoryExtractionJobAttributionMode;
costCents: number;
resultSummary: string | null;
errorCode: string | null;
error: string | null;
retryOfJobId: string | null;
attemptNumber: number;
retryCount: number;
}
export interface MemoryExtractionJobDetail extends MemoryExtractionJob {
effectiveState: MemoryExtractionJobEffectiveState;
retryCount: number;
rerunEligible: boolean;
}
export interface MemoryExtractionJobListResponse {
jobs: MemoryExtractionJobListItem[];
nextOffset: number | null;
}
export interface MemoryExtractionJobDetailResponse {
job: MemoryExtractionJobDetail;
}
export interface MemoryExtractionJobRerunResponse {
job: MemoryExtractionJobDetail;
}
+14
View File
@@ -246,6 +246,20 @@ export {
type CreateFinanceEvent,
} from "./finance.js";
export {
listMemoryExtractionJobsQuerySchema,
memoryExtractionJobAttributionModeSchema,
memoryExtractionJobDispatcherKindSchema,
memoryExtractionJobEffectiveStateSchema,
memoryExtractionJobHookKindSchema,
memoryExtractionJobOperationTypeSchema,
memoryExtractionJobSourceKindSchema,
memoryExtractionJobStatusSchema,
rerunMemoryExtractionJobSchema,
type ListMemoryExtractionJobsQuery,
type RerunMemoryExtractionJob,
} from "./memory.js";
export {
createAssetImageMetadataSchema,
type CreateAssetImageMetadata,
+56
View File
@@ -0,0 +1,56 @@
import { z } from "zod";
import {
MEMORY_EXTRACTION_JOB_ATTRIBUTION_MODES,
MEMORY_EXTRACTION_JOB_DISPATCHER_KINDS,
MEMORY_EXTRACTION_JOB_EFFECTIVE_STATES,
MEMORY_EXTRACTION_JOB_HOOK_KINDS,
MEMORY_EXTRACTION_JOB_OPERATION_TYPES,
MEMORY_EXTRACTION_JOB_SOURCE_KINDS,
MEMORY_EXTRACTION_JOB_STATUSES,
} from "../constants.js";
const memoryBindingKeySchema = z
.string()
.trim()
.min(1)
.max(64)
.regex(/^[a-z0-9][a-z0-9_-]*$/, "Binding key must be lowercase letters, numbers, _ or -");
export const memoryExtractionJobOperationTypeSchema = z.enum(MEMORY_EXTRACTION_JOB_OPERATION_TYPES);
export const memoryExtractionJobStatusSchema = z.enum(MEMORY_EXTRACTION_JOB_STATUSES);
export const memoryExtractionJobHookKindSchema = z.enum(MEMORY_EXTRACTION_JOB_HOOK_KINDS);
export const memoryExtractionJobSourceKindSchema = z.enum(MEMORY_EXTRACTION_JOB_SOURCE_KINDS);
export const memoryExtractionJobDispatcherKindSchema = z.enum(MEMORY_EXTRACTION_JOB_DISPATCHER_KINDS);
export const memoryExtractionJobAttributionModeSchema = z.enum(MEMORY_EXTRACTION_JOB_ATTRIBUTION_MODES);
export const memoryExtractionJobEffectiveStateSchema = z.enum(MEMORY_EXTRACTION_JOB_EFFECTIVE_STATES);
export const listMemoryExtractionJobsQuerySchema = z
.object({
status: memoryExtractionJobStatusSchema.optional(),
effectiveState: memoryExtractionJobEffectiveStateSchema.optional(),
bindingKey: memoryBindingKeySchema.optional(),
operationType: memoryExtractionJobOperationTypeSchema.optional(),
agentId: z.string().uuid().optional(),
issueId: z.string().uuid().optional(),
runId: z.string().uuid().optional(),
submittedAfter: z.coerce.date().optional(),
submittedBefore: z.coerce.date().optional(),
limit: z.coerce.number().int().positive().max(200).optional().default(50),
offset: z.coerce.number().int().min(0).optional().default(0),
})
.strict()
.superRefine((value, ctx) => {
if (value.submittedAfter && value.submittedBefore && value.submittedAfter > value.submittedBefore) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
path: ["submittedAfter"],
message: "submittedAfter must be before or equal to submittedBefore",
});
}
});
export type ListMemoryExtractionJobsQuery = z.infer<typeof listMemoryExtractionJobsQuerySchema>;
export const rerunMemoryExtractionJobSchema = z.object({}).strict();
export type RerunMemoryExtractionJob = z.infer<typeof rerunMemoryExtractionJobSchema>;
@@ -0,0 +1,147 @@
import fs from "node:fs";
import net from "node:net";
import os from "node:os";
import path from "node:path";
import EmbeddedPostgres from "embedded-postgres";
import { createDb, ensurePostgresDatabase } from "@paperclipai/db";
export type MemoryJobTestDatabase = {
connectionString: string;
cleanup(): Promise<void>;
};
const MEMORY_JOB_TEST_SCHEMA_SQL = `
CREATE EXTENSION IF NOT EXISTS pgcrypto;
CREATE TABLE companies (
id uuid PRIMARY KEY,
name text NOT NULL,
description text,
status text NOT NULL DEFAULT 'active',
pause_reason text,
paused_at timestamp with time zone,
issue_prefix text NOT NULL,
issue_counter integer NOT NULL DEFAULT 0,
budget_monthly_cents integer NOT NULL DEFAULT 0,
spent_monthly_cents integer NOT NULL DEFAULT 0,
require_board_approval_for_new_agents boolean NOT NULL DEFAULT false,
feedback_data_sharing_enabled boolean NOT NULL DEFAULT false,
feedback_data_sharing_consent_at timestamp with time zone,
feedback_data_sharing_consent_by_user_id text,
feedback_data_sharing_terms_version text,
brand_color text,
created_at timestamp with time zone NOT NULL DEFAULT now(),
updated_at timestamp with time zone NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX companies_issue_prefix_idx ON companies (issue_prefix);
CREATE TABLE memory_extraction_jobs (
id uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
company_id uuid NOT NULL,
binding_id uuid NOT NULL,
binding_key text NOT NULL,
operation_type text NOT NULL,
status text DEFAULT 'queued' NOT NULL,
source_agent_id uuid,
source_issue_id uuid,
source_project_id uuid,
source_goal_id uuid,
source_heartbeat_run_id uuid,
hook_kind text,
provider_job_id text,
submitted_at timestamp with time zone DEFAULT now() NOT NULL,
started_at timestamp with time zone,
finished_at timestamp with time zone,
attribution_mode text DEFAULT 'untracked' NOT NULL,
cost_cents integer DEFAULT 0 NOT NULL,
result_summary text,
error_code text,
error text,
source_kind text NOT NULL,
source_ref_json jsonb,
retry_of_job_id uuid,
attempt_number integer DEFAULT 1 NOT NULL,
dispatcher_kind text DEFAULT 'in_process' NOT NULL,
lease_expires_at timestamp with time zone,
usage_json jsonb,
result_json jsonb,
created_at timestamp with time zone DEFAULT now() NOT NULL,
updated_at timestamp with time zone DEFAULT now() NOT NULL,
CONSTRAINT memory_extraction_jobs_company_id_companies_id_fk FOREIGN KEY (company_id) REFERENCES companies(id)
);
CREATE INDEX memory_extraction_jobs_company_status_submitted_idx
ON memory_extraction_jobs (company_id, status, submitted_at);
CREATE INDEX memory_extraction_jobs_company_status_lease_expires_idx
ON memory_extraction_jobs (company_id, status, lease_expires_at);
CREATE INDEX memory_extraction_jobs_retry_of_job_idx
ON memory_extraction_jobs (retry_of_job_id);
`;
async function getAvailablePort(): Promise<number> {
return await new Promise((resolve, reject) => {
const server = net.createServer();
server.unref();
server.on("error", reject);
server.listen(0, "127.0.0.1", () => {
const address = server.address();
if (!address || typeof address === "string") {
server.close(() => reject(new Error("Failed to allocate memory-job test port")));
return;
}
const { port } = address;
server.close((error) => {
if (error) reject(error);
else resolve(port);
});
});
});
}
export async function startMemoryJobTestDatabase(
tempDirPrefix: string,
): Promise<MemoryJobTestDatabase> {
const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), tempDirPrefix));
const port = await getAvailablePort();
const instance = new EmbeddedPostgres({
databaseDir: dataDir,
user: "paperclip",
password: "paperclip",
port,
persistent: true,
initdbFlags: ["--encoding=UTF8", "--locale=C", "--lc-messages=C"],
onLog: () => {},
onError: () => {},
});
try {
await instance.initialise();
await instance.start();
const adminConnectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/postgres`;
await ensurePostgresDatabase(adminConnectionString, "paperclip");
const connectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/paperclip`;
const db = createDb(connectionString) as unknown as {
$client: {
unsafe(query: string): Promise<unknown>;
end(): Promise<unknown>;
};
};
await db.$client.unsafe(MEMORY_JOB_TEST_SCHEMA_SQL);
await db.$client.end();
return {
connectionString,
cleanup: async () => {
await instance.stop().catch(() => {});
fs.rmSync(dataDir, { recursive: true, force: true });
},
};
} catch (error) {
await instance.stop().catch(() => {});
fs.rmSync(dataDir, { recursive: true, force: true });
throw error;
}
}
@@ -0,0 +1,231 @@
import { randomUUID } from "node:crypto";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
companies,
createDb,
memoryExtractionJobs,
} from "@paperclipai/db";
import {
getEmbeddedPostgresTestSupport,
} from "./helpers/embedded-postgres.js";
import { startMemoryJobTestDatabase } from "./helpers/memory-job-test-db.ts";
import {
buildSyntheticMemoryBindingId,
createPostRunMemoryCaptureHandler,
enqueuePostRunMemoryCaptureJob,
} from "../services/memory-job-capture.ts";
import { createMemoryJobDispatcher } from "../services/memory-job-dispatcher.ts";
import { memoryJobStore } from "../services/memory-job-store.ts";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres memory job capture tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
}
async function insertCompany(db: ReturnType<typeof createDb>, companyId: string) {
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
}
describeEmbeddedPostgres("memory job capture", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startMemoryJobTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startMemoryJobTestDatabase("paperclip-memory-job-capture-");
db = createDb(tempDb.connectionString);
}, 20_000);
afterEach(async () => {
await db.delete(memoryExtractionJobs);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
it("enqueues issue-scoped post-run capture jobs with provenance and included run cost", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const projectId = randomUUID();
const goalId = randomUUID();
await insertCompany(db, companyId);
const store = memoryJobStore(db);
const queued = await enqueuePostRunMemoryCaptureJob({
store,
run: {
id: randomUUID(),
companyId,
agentId,
status: "succeeded",
invocationSource: "heartbeat",
triggerDetail: "manual",
usageJson: {
provider: "codex",
biller: "openai",
model: "o4-mini",
billingType: "metered_api",
inputTokens: 120,
cachedInputTokens: 30,
outputTokens: 42,
latencyMs: 88,
costUsd: 0.12,
rawInputTokens: 144,
},
resultJson: {
summary: "Implemented the requested memory-job wiring.",
total_cost_usd: 0.12,
},
error: null,
errorCode: null,
sessionIdAfter: "codex-session-1",
},
sourceIssueId: issueId,
sourceProjectId: projectId,
sourceGoalId: goalId,
});
expect(queued).toMatchObject({
companyId,
bindingId: buildSyntheticMemoryBindingId(companyId, "primary"),
bindingKey: "primary",
operationType: "capture",
status: "queued",
sourceKind: "run",
sourceAgentId: agentId,
sourceIssueId: issueId,
sourceProjectId: projectId,
sourceGoalId: goalId,
hookKind: "post_run_capture",
attributionMode: "included_in_run",
costCents: 12,
resultSummary: "Implemented the requested memory-job wiring.",
});
expect(queued.usageJson).toMatchObject({
provider: "codex",
biller: "openai",
model: "o4-mini",
billingType: "metered_api",
inputTokens: 120,
cachedInputTokens: 30,
outputTokens: 42,
latencyMs: 88,
costCents: 12,
details: {
rawInputTokens: 144,
},
});
expect(queued.sourceRefJson).toMatchObject({
invocationSource: "heartbeat",
triggerDetail: "manual",
sourceRunStatus: "succeeded",
sessionIdAfter: "codex-session-1",
});
expect(queued.resultJson).toMatchObject({
captureState: "queued",
hookKind: "post_run_capture",
sourceKind: "run",
sourceRunId: queued.sourceHeartbeatRunId,
sourceRunStatus: "succeeded",
sourceSessionIdAfter: "codex-session-1",
sourceResult: {
summary: "Implemented the requested memory-job wiring.",
total_cost_usd: 0.12,
},
});
});
it("supports direct-billed capture overrides and reruns capture-originated jobs", async () => {
const companyId = randomUUID();
const runId = randomUUID();
await insertCompany(db, companyId);
const store = memoryJobStore(db);
const queued = await enqueuePostRunMemoryCaptureJob({
store,
run: {
id: runId,
companyId,
agentId: randomUUID(),
status: "succeeded",
invocationSource: "heartbeat",
triggerDetail: "issue_assigned",
usageJson: null,
resultJson: {
summary: "Posted a concise issue handoff.",
},
error: null,
errorCode: null,
sessionIdAfter: null,
},
sourceIssueId: randomUUID(),
sourceProjectId: randomUUID(),
sourceGoalId: randomUUID(),
attributionMode: "billed_directly",
costCents: 31,
usageJson: {
provider: "memory-provider",
biller: "memory-provider",
model: "capture-v1",
billingType: "metered_api",
inputTokens: 11,
outputTokens: 5,
costCents: 31,
latencyMs: 54,
},
});
const dispatcher = createMemoryJobDispatcher({
store,
resolveHandler: () => createPostRunMemoryCaptureHandler(),
});
await dispatcher.tick();
const completed = await store.getDetail(companyId, queued.id);
expect(completed).toMatchObject({
id: queued.id,
status: "succeeded",
effectiveState: "succeeded",
attributionMode: "billed_directly",
costCents: 31,
resultSummary: "Posted a concise issue handoff.",
});
expect(completed?.usageJson).toMatchObject({
provider: "memory-provider",
model: "capture-v1",
costCents: 31,
});
expect(completed?.resultJson).toMatchObject({
captureState: "succeeded",
sourceRunId: runId,
persistedUsageSummary: true,
});
const rerun = await store.rerun(companyId, queued.id, {
now: new Date("2026-04-21T22:00:00.000Z"),
});
expect(rerun).toMatchObject({
status: "queued",
retryOfJobId: queued.id,
attemptNumber: 2,
sourceKind: "run",
hookKind: "post_run_capture",
sourceHeartbeatRunId: runId,
bindingKey: "primary",
attributionMode: "billed_directly",
});
});
});
@@ -0,0 +1,166 @@
import { randomUUID } from "node:crypto";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
companies,
createDb,
memoryExtractionJobs,
} from "@paperclipai/db";
import {
getEmbeddedPostgresTestSupport,
} from "./helpers/embedded-postgres.js";
import { startMemoryJobTestDatabase } from "./helpers/memory-job-test-db.ts";
import { createMemoryJobDispatcher } from "../services/memory-job-dispatcher.ts";
import { memoryJobStore } from "../services/memory-job-store.ts";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres memory job dispatcher tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
}
async function insertCompany(db: ReturnType<typeof createDb>, companyId: string) {
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
}
describeEmbeddedPostgres("memory job dispatcher", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startMemoryJobTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startMemoryJobTestDatabase("paperclip-memory-job-dispatcher-");
db = createDb(tempDb.connectionString);
}, 20_000);
afterEach(async () => {
await db.delete(memoryExtractionJobs);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
it("claims queued jobs and completes them with handler output", async () => {
const companyId = randomUUID();
let currentTime = new Date("2026-04-21T20:00:00.000Z");
await insertCompany(db, companyId);
const store = memoryJobStore(db);
const queued = await store.enqueue({
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
sourceKind: "manual",
});
const dispatcher = createMemoryJobDispatcher({
store,
now: () => currentTime,
resolveHandler: () => async () => ({
outcome: "succeeded",
providerJobId: "provider-job-1",
resultSummary: "Captured 3 records",
costCents: 7,
usageJson: {
provider: "test-provider",
model: "test-model",
inputTokens: 10,
outputTokens: 4,
latencyMs: 55,
},
resultJson: { recordsCaptured: 3 },
}),
});
await dispatcher.tick();
const detail = await store.getDetail(companyId, queued.id, { now: currentTime });
expect(detail).toMatchObject({
id: queued.id,
status: "succeeded",
effectiveState: "succeeded",
providerJobId: "provider-job-1",
resultSummary: "Captured 3 records",
costCents: 7,
});
expect(detail?.usageJson).toMatchObject({
provider: "test-provider",
model: "test-model",
inputTokens: 10,
outputTokens: 4,
latencyMs: 55,
});
expect(detail?.resultJson).toEqual({ recordsCaptured: 3 });
});
it("keeps provider-managed jobs running until the recovery sweep expires their lease", async () => {
const companyId = randomUUID();
let currentTime = new Date("2026-04-21T21:00:00.000Z");
await insertCompany(db, companyId);
const store = memoryJobStore(db);
const queued = await store.enqueue({
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
sourceKind: "manual",
});
const dispatcher = createMemoryJobDispatcher({
store,
now: () => currentTime,
leaseDurationMs: 5_000,
resolveHandler: () => async () => ({
outcome: "running",
providerJobId: "provider-job-async",
resultSummary: "Submitted to provider",
leaseDurationMs: 1_000,
resultJson: { submitted: true },
}),
});
await dispatcher.tick();
const runningDetail = await store.getDetail(companyId, queued.id, { now: currentTime });
expect(runningDetail).toMatchObject({
id: queued.id,
status: "running",
effectiveState: "running",
providerJobId: "provider-job-async",
resultSummary: "Submitted to provider",
});
currentTime = new Date("2026-04-21T21:00:02.000Z");
const staleDetail = await store.getDetail(companyId, queued.id, { now: currentTime });
expect(staleDetail).toMatchObject({
id: queued.id,
status: "running",
effectiveState: "stuck",
});
const recovery = await dispatcher.sweepExpiredLeases();
const recoveredDetail = await store.getDetail(companyId, queued.id, { now: currentTime });
expect(recovery).toEqual({ recovered: 1, jobIds: [queued.id] });
expect(recoveredDetail).toMatchObject({
id: queued.id,
status: "failed",
effectiveState: "failed",
errorCode: "lease_expired",
error: "Memory job lease expired before completion",
providerJobId: "provider-job-async",
});
});
});
@@ -0,0 +1,241 @@
import { randomUUID } from "node:crypto";
import express from "express";
import request from "supertest";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import {
companies,
createDb,
memoryExtractionJobs,
} from "@paperclipai/db";
import {
getEmbeddedPostgresTestSupport,
} from "./helpers/embedded-postgres.js";
import { startMemoryJobTestDatabase } from "./helpers/memory-job-test-db.ts";
import { errorHandler } from "../middleware/index.js";
import { memoryJobRoutes } from "../routes/memory-jobs.js";
const logActivityMock = vi.hoisted(() => vi.fn());
vi.mock("../services/index.js", () => ({
logActivity: logActivityMock,
}));
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres memory job route tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
}
async function insertCompany(db: ReturnType<typeof createDb>, companyId: string) {
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
}
function createApp(db: ReturnType<typeof createDb>, actor: Record<string, unknown>) {
const app = express();
app.use(express.json());
app.use((req, _res, next) => {
req.actor = actor as never;
next();
});
app.use("/api", memoryJobRoutes(db as never));
app.use(errorHandler);
return app;
}
describeEmbeddedPostgres("memory job routes", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startMemoryJobTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startMemoryJobTestDatabase("paperclip-memory-job-routes-");
db = createDb(tempDb.connectionString);
}, 20_000);
beforeEach(() => {
logActivityMock.mockReset();
logActivityMock.mockResolvedValue(undefined);
});
afterEach(async () => {
await db.delete(memoryExtractionJobs);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
it("lists company-scoped jobs with derived stuck-state filters", async () => {
const companyId = randomUUID();
const otherCompanyId = randomUUID();
const stuckJobId = randomUUID();
const otherJobId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const runId = randomUUID();
await insertCompany(db, companyId);
await insertCompany(db, otherCompanyId);
await db.insert(memoryExtractionJobs).values([
{
id: stuckJobId,
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
status: "running",
sourceKind: "manual",
sourceAgentId: agentId,
sourceIssueId: issueId,
sourceHeartbeatRunId: runId,
submittedAt: new Date("2026-04-21T18:00:00.000Z"),
startedAt: new Date("2026-04-21T18:01:00.000Z"),
leaseExpiresAt: new Date("2026-04-21T18:05:00.000Z"),
},
{
id: otherJobId,
companyId: otherCompanyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
status: "running",
sourceKind: "manual",
sourceAgentId: agentId,
sourceIssueId: issueId,
sourceHeartbeatRunId: runId,
submittedAt: new Date("2026-04-21T18:10:00.000Z"),
startedAt: new Date("2026-04-21T18:11:00.000Z"),
leaseExpiresAt: new Date("2026-04-21T18:12:00.000Z"),
},
]);
const app = createApp(db, {
type: "board",
userId: "board",
source: "local_implicit",
isInstanceAdmin: true,
});
const res = await request(app).get(
`/api/companies/${companyId}/memory/jobs?effectiveState=stuck&bindingKey=primary&operationType=capture&agentId=${agentId}&issueId=${issueId}&runId=${runId}`,
);
expect(res.status).toBe(200);
expect(res.body.nextOffset).toBeNull();
expect(res.body.jobs).toHaveLength(1);
expect(res.body.jobs[0]).toMatchObject({
id: stuckJobId,
effectiveState: "stuck",
sourceAgentId: agentId,
sourceIssueId: issueId,
sourceHeartbeatRunId: runId,
});
});
it("returns job detail scoped to the requested company", async () => {
const companyId = randomUUID();
const jobId = randomUUID();
await insertCompany(db, companyId);
await db.insert(memoryExtractionJobs).values({
id: jobId,
companyId,
bindingId: randomUUID(),
bindingKey: "capture-main",
operationType: "capture",
status: "failed",
sourceKind: "manual",
submittedAt: new Date("2026-04-21T18:00:00.000Z"),
finishedAt: new Date("2026-04-21T18:03:00.000Z"),
errorCode: "provider_error",
error: "Provider timed out",
});
const app = createApp(db, {
type: "board",
userId: "board",
source: "local_implicit",
isInstanceAdmin: true,
});
const res = await request(app).get(`/api/companies/${companyId}/memory/jobs/${jobId}`);
expect(res.status).toBe(200);
expect(res.body.job).toMatchObject({
id: jobId,
bindingKey: "capture-main",
status: "failed",
effectiveState: "failed",
errorCode: "provider_error",
rerunEligible: true,
});
});
it("reruns an eligible job, preserves lineage, and logs activity", async () => {
const companyId = randomUUID();
const jobId = randomUUID();
await insertCompany(db, companyId);
await db.insert(memoryExtractionJobs).values({
id: jobId,
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
status: "failed",
sourceKind: "manual",
submittedAt: new Date("2026-04-21T18:00:00.000Z"),
finishedAt: new Date("2026-04-21T18:02:00.000Z"),
costCents: 41,
resultSummary: "Provider failed",
errorCode: "provider_error",
error: "Provider timed out",
});
const app = createApp(db, {
type: "board",
userId: "board",
source: "local_implicit",
isInstanceAdmin: true,
});
const res = await request(app)
.post(`/api/companies/${companyId}/memory/jobs/${jobId}/rerun`)
.send({});
expect(res.status).toBe(201);
expect(res.body.job).toMatchObject({
status: "queued",
bindingKey: "primary",
retryOfJobId: jobId,
attemptNumber: 2,
costCents: 0,
resultSummary: null,
errorCode: null,
error: null,
rerunEligible: false,
});
expect(logActivityMock).toHaveBeenCalledWith(
db,
expect.objectContaining({
companyId,
action: "memory_job_rerun",
entityType: "memory_job",
entityId: res.body.job.id,
details: expect.objectContaining({
sourceJobId: jobId,
retryOfJobId: jobId,
attemptNumber: 2,
}),
}),
);
});
});
@@ -0,0 +1,158 @@
import { randomUUID } from "node:crypto";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
companies,
createDb,
memoryExtractionJobs,
} from "@paperclipai/db";
import {
getEmbeddedPostgresTestSupport,
} from "./helpers/embedded-postgres.js";
import { startMemoryJobTestDatabase } from "./helpers/memory-job-test-db.ts";
import { memoryJobStore } from "../services/memory-job-store.ts";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres memory job store tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
}
async function insertCompany(db: ReturnType<typeof createDb>, companyId: string) {
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
}
describeEmbeddedPostgres("memory job store", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startMemoryJobTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startMemoryJobTestDatabase("paperclip-memory-job-store-");
db = createDb(tempDb.connectionString);
}, 20_000);
afterEach(async () => {
await db.delete(memoryExtractionJobs);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
it("creates rerun attempts against the original lineage without mutating prior attempts", async () => {
const companyId = randomUUID();
const store = memoryJobStore(db);
await insertCompany(db, companyId);
const original = await store.enqueue({
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
sourceKind: "manual",
});
await store.cancel({
companyId,
jobId: original.id,
now: new Date("2026-04-21T18:00:00.000Z"),
resultSummary: "Cancelled by test",
});
const firstRetry = await store.rerun(companyId, original.id, {
now: new Date("2026-04-21T18:01:00.000Z"),
});
await store.cancel({
companyId,
jobId: firstRetry.id,
now: new Date("2026-04-21T18:02:00.000Z"),
resultSummary: "Retry cancelled by test",
});
const secondRetry = await store.rerun(companyId, firstRetry.id, {
now: new Date("2026-04-21T18:03:00.000Z"),
});
const originalDetail = await store.getDetail(companyId, original.id);
const secondRetryDetail = await store.getDetail(companyId, secondRetry.id);
expect(originalDetail).toMatchObject({
id: original.id,
status: "cancelled",
attemptNumber: 1,
retryCount: 2,
rerunEligible: true,
});
expect(firstRetry).toMatchObject({
retryOfJobId: original.id,
attemptNumber: 2,
status: "queued",
});
expect(secondRetryDetail).toMatchObject({
id: secondRetry.id,
retryOfJobId: original.id,
attemptNumber: 3,
status: "queued",
retryCount: 2,
rerunEligible: false,
});
});
it("derives stuck effective state from expired running leases in list and detail queries", async () => {
const companyId = randomUUID();
const now = new Date("2026-04-21T19:00:00.000Z");
const staleJobId = randomUUID();
const activeJobId = randomUUID();
await insertCompany(db, companyId);
await db.insert(memoryExtractionJobs).values([
{
id: staleJobId,
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
status: "running",
sourceKind: "manual",
startedAt: new Date("2026-04-21T18:40:00.000Z"),
submittedAt: new Date("2026-04-21T18:39:00.000Z"),
leaseExpiresAt: new Date("2026-04-21T18:50:00.000Z"),
},
{
id: activeJobId,
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
status: "running",
sourceKind: "manual",
startedAt: new Date("2026-04-21T18:55:00.000Z"),
submittedAt: new Date("2026-04-21T18:54:00.000Z"),
leaseExpiresAt: new Date("2026-04-21T19:05:00.000Z"),
},
]);
const store = memoryJobStore(db);
const stuckList = await store.list(companyId, { effectiveState: "stuck" }, { now });
const runningList = await store.list(companyId, { effectiveState: "running" }, { now });
const staleDetail = await store.getDetail(companyId, staleJobId, { now });
expect(stuckList.jobs.map((job) => job.id)).toEqual([staleJobId]);
expect(runningList.jobs.map((job) => job.id)).toEqual([activeJobId]);
expect(staleDetail).toMatchObject({
id: staleJobId,
status: "running",
effectiveState: "stuck",
rerunEligible: false,
});
});
});
+2
View File
@@ -22,6 +22,7 @@ import { approvalRoutes } from "./routes/approvals.js";
import { secretRoutes } from "./routes/secrets.js";
import { costRoutes } from "./routes/costs.js";
import { activityRoutes } from "./routes/activity.js";
import { memoryJobRoutes } from "./routes/memory-jobs.js";
import { dashboardRoutes } from "./routes/dashboard.js";
import { userProfileRoutes } from "./routes/user-profiles.js";
import { sidebarBadgeRoutes } from "./routes/sidebar-badges.js";
@@ -196,6 +197,7 @@ export async function createApp(
api.use(secretRoutes(db));
api.use(costRoutes(db));
api.use(activityRoutes(db));
api.use(memoryJobRoutes(db));
api.use(dashboardRoutes(db));
api.use(userProfileRoutes(db));
api.use(sidebarBadgeRoutes(db));
+16
View File
@@ -42,6 +42,9 @@ import { getBoardClaimWarningUrl, initializeBoardClaimChallenge } from "./board-
import { maybePersistWorktreeRuntimePorts } from "./worktree-config.js";
import { initTelemetry, getTelemetryClient } from "./telemetry.js";
import { conflict } from "./errors.js";
import { createMemoryJobDispatcher } from "./services/memory-job-dispatcher.js";
import { createPostRunMemoryCaptureHandler } from "./services/memory-job-capture.js";
import { memoryJobStore } from "./services/memory-job-store.js";
import type {
InstanceDatabaseBackupRunResult,
InstanceDatabaseBackupTrigger,
@@ -642,6 +645,18 @@ export async function startServer(): Promise<StartedServer> {
resolveSessionFromHeaders,
});
const postRunMemoryCaptureHandler = createPostRunMemoryCaptureHandler();
const memoryJobDispatcher = createMemoryJobDispatcher({
store: memoryJobStore(db as any),
resolveHandler: (job) => {
if (job.operationType === "capture" && job.sourceKind === "run" && job.hookKind === "post_run_capture") {
return postRunMemoryCaptureHandler;
}
return null;
},
});
memoryJobDispatcher.start();
void reconcilePersistedRuntimeServicesOnStartup(db as any)
.then((result) => {
if (result.reconciled > 0) {
@@ -831,6 +846,7 @@ export async function startServer(): Promise<StartedServer> {
{
const shutdown = async (signal: "SIGINT" | "SIGTERM") => {
const telemetryClient = getTelemetryClient();
memoryJobDispatcher.stop();
if (telemetryClient) {
telemetryClient.stop();
await telemetryClient.flush();
+77
View File
@@ -0,0 +1,77 @@
import { Router } from "express";
import type { Db } from "@paperclipai/db";
import {
listMemoryExtractionJobsQuerySchema,
rerunMemoryExtractionJobSchema,
} from "@paperclipai/shared";
import { notFound } from "../errors.js";
import { validate } from "../middleware/validate.js";
import { logActivity } from "../services/index.js";
import { memoryJobStore } from "../services/memory-job-store.js";
import { assertCompanyAccess, getActorInfo } from "./authz.js";
export function memoryJobRoutes(db: Db) {
const router = Router();
const store = memoryJobStore(db);
router.get("/companies/:companyId/memory/jobs", async (req, res) => {
const companyId = req.params.companyId as string;
assertCompanyAccess(req, companyId);
const query = listMemoryExtractionJobsQuerySchema.parse(req.query);
const result = await store.list(companyId, query);
res.json(result);
});
router.get("/companies/:companyId/memory/jobs/:jobId", async (req, res) => {
const companyId = req.params.companyId as string;
const jobId = req.params.jobId as string;
assertCompanyAccess(req, companyId);
const job = await store.getDetail(companyId, jobId);
if (!job) {
throw notFound(`Memory job not found: ${jobId}`);
}
res.json({ job });
});
router.post(
"/companies/:companyId/memory/jobs/:jobId/rerun",
validate(rerunMemoryExtractionJobSchema),
async (req, res) => {
const companyId = req.params.companyId as string;
const jobId = req.params.jobId as string;
assertCompanyAccess(req, companyId);
const actor = getActorInfo(req);
const rerunJob = await store.rerun(companyId, jobId);
const job = await store.getDetail(companyId, rerunJob.id);
if (!job) {
throw notFound(`Memory job not found: ${rerunJob.id}`);
}
await logActivity(db, {
companyId,
actorType: actor.actorType,
actorId: actor.actorId,
agentId: actor.agentId,
runId: actor.runId,
action: "memory_job_rerun",
entityType: "memory_job",
entityId: job.id,
details: {
sourceJobId: jobId,
bindingKey: job.bindingKey,
operationType: job.operationType,
retryOfJobId: job.retryOfJobId,
attemptNumber: job.attemptNumber,
},
});
res.status(201).json({ job });
},
);
return router;
}
+35
View File
@@ -83,6 +83,8 @@ import {
getIssueContinuationSummaryDocument,
refreshIssueContinuationSummary,
} from "./issue-continuation-summary.js";
import { enqueuePostRunMemoryCaptureJob } from "./memory-job-capture.js";
import { memoryJobStore } from "./memory-job-store.js";
import { executionWorkspaceService, mergeExecutionWorkspaceConfig } from "./execution-workspaces.js";
import { workspaceOperationService } from "./workspace-operations.js";
import { isProcessGroupAlive, terminateLocalService } from "./local-service-supervisor.js";
@@ -1822,6 +1824,7 @@ export function heartbeatService(db: Db) {
const secretsSvc = secretService(db);
const companySkills = companySkillService(db);
const issuesSvc = issueService(db);
const memoryJobs = memoryJobStore(db);
const executionWorkspacesSvc = executionWorkspaceService(db);
const workspaceOperationsSvc = workspaceOperationService(db);
const activeRunExecutions = new Set<string>();
@@ -1902,6 +1905,17 @@ export function heartbeatService(db: Db) {
.then((rows) => rows[0] ?? null);
}
async function getIssueMemoryCaptureScope(companyId: string, issueId: string) {
return db
.select({
projectId: issues.projectId,
goalId: issues.goalId,
})
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.id, issueId)))
.then((rows) => rows[0] ?? null);
}
async function getRuntimeState(agentId: string) {
return db
.select()
@@ -5665,6 +5679,27 @@ export function heartbeatService(db: Db) {
`[paperclip] Failed to post run summary comment: ${err instanceof Error ? err.message : String(err)}\n`,
);
}
try {
const contextSnapshot = parseObject(livenessRun.contextSnapshot);
const issueScope = await getIssueMemoryCaptureScope(livenessRun.companyId, issueId);
await enqueuePostRunMemoryCaptureJob({
store: memoryJobs,
run: livenessRun,
sourceIssueId: issueId,
sourceProjectId: issueScope?.projectId ?? readNonEmptyString(contextSnapshot.projectId),
sourceGoalId: issueScope?.goalId ?? readNonEmptyString(contextSnapshot.goalId),
});
} catch (err) {
logger.warn(
{
error: err instanceof Error ? err.message : String(err),
runId: livenessRun.id,
issueId,
},
"failed to enqueue post-run memory capture job",
);
}
}
if (outcome === "failed" && livenessRun.errorCode === "codex_transient_upstream") {
await scheduleBoundedRetryForRun(livenessRun, agent);
+331
View File
@@ -0,0 +1,331 @@
import { createHash } from "node:crypto";
import type {
BillingType,
MemoryExtractionJobAttributionMode,
MemoryExtractionJobDispatcherKind,
MemoryExtractionJobSourceRef,
MemoryExtractionJobUsage,
} from "@paperclipai/shared";
import {
buildHeartbeatRunIssueComment,
summarizeHeartbeatRunResultJson,
} from "./heartbeat-run-summary.js";
import type { MemoryJobHandler } from "./memory-job-dispatcher.js";
import type { MemoryJobStore } from "./memory-job-store.js";
export const DEFAULT_POST_RUN_MEMORY_BINDING_KEY = "primary";
type PostRunMemoryCaptureSource = {
id: string;
companyId: string;
agentId: string;
status: string;
invocationSource: string;
triggerDetail: string | null;
usageJson: Record<string, unknown> | null;
resultJson: Record<string, unknown> | null;
error: string | null;
errorCode: string | null;
sessionIdAfter: string | null;
};
export interface EnqueuePostRunMemoryCaptureInput {
store: MemoryJobStore;
run: PostRunMemoryCaptureSource;
sourceIssueId?: string | null;
sourceProjectId?: string | null;
sourceGoalId?: string | null;
bindingKey?: string;
bindingId?: string;
attributionMode?: MemoryExtractionJobAttributionMode;
costCents?: number;
usageJson?: MemoryExtractionJobUsage | null;
resultSummary?: string | null;
resultJson?: Record<string, unknown> | null;
sourceRefJson?: MemoryExtractionJobSourceRef | null;
dispatcherKind?: MemoryExtractionJobDispatcherKind;
}
function asRecord(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: null;
}
function readNonEmptyString(value: unknown): string | null {
if (typeof value !== "string") {
return null;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : null;
}
function readFiniteNumber(value: unknown): number | null {
return typeof value === "number" && Number.isFinite(value) ? value : null;
}
function readFiniteField(record: Record<string, unknown> | null, key: string): number | null {
return record ? readFiniteNumber(record[key]) : null;
}
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)}`;
}
export function buildSyntheticMemoryBindingId(companyId: string, bindingKey: string): string {
const digest = createHash("sha256")
.update(`memory-binding:${companyId}:${bindingKey}`)
.digest("hex");
return formatUuidFromHex(digest);
}
function normalizeBillingType(value: unknown): BillingType | null {
const raw = readNonEmptyString(value);
switch (raw) {
case "metered_api":
case "api":
return "metered_api";
case "subscription_included":
case "subscription":
return "subscription_included";
case "subscription_overage":
return "subscription_overage";
case "credits":
return "credits";
case "fixed":
return "fixed";
case "unknown":
return "unknown";
default:
return null;
}
}
function costCentsFromUsage(usageJson: MemoryExtractionJobUsage | null): number | null {
if (!usageJson) {
return null;
}
if (typeof usageJson.costCents === "number" && Number.isFinite(usageJson.costCents)) {
return Math.max(0, Math.round(usageJson.costCents));
}
return null;
}
function buildUsageSummaryFromRun(
usageJson: Record<string, unknown> | null,
): MemoryExtractionJobUsage | null {
const usage = asRecord(usageJson);
if (!usage) {
return null;
}
const billingType = normalizeBillingType(usage.billingType);
const rawCostCents = readFiniteField(usage, "costCents");
const rawCostUsd = readFiniteField(usage, "costUsd");
const derivedCostCents =
rawCostCents !== null
? Math.max(0, Math.round(rawCostCents))
: rawCostUsd !== null
? Math.max(0, Math.round(rawCostUsd * 100))
: null;
const details = Object.fromEntries(
Object.entries(usage).filter(([key]) => ![
"provider",
"biller",
"model",
"billingType",
"inputTokens",
"cachedInputTokens",
"outputTokens",
"embeddingTokens",
"costCents",
"costUsd",
"latencyMs",
].includes(key)),
);
if (
Object.keys(usage).length === 0
&& Object.keys(details).length === 0
) {
return null;
}
return {
provider: readNonEmptyString(usage.provider) ?? "unknown",
biller: readNonEmptyString(usage.biller),
model: readNonEmptyString(usage.model),
billingType,
inputTokens: readFiniteField(usage, "inputTokens"),
cachedInputTokens: readFiniteField(usage, "cachedInputTokens"),
outputTokens: readFiniteField(usage, "outputTokens"),
embeddingTokens: readFiniteField(usage, "embeddingTokens"),
costCents: derivedCostCents,
latencyMs: readFiniteField(usage, "latencyMs"),
details: Object.keys(details).length > 0 ? details : null,
};
}
function resolveAttribution(input: {
attributionMode?: MemoryExtractionJobAttributionMode;
costCents?: number;
usageJson: MemoryExtractionJobUsage | null;
}) {
const derivedCostCents = costCentsFromUsage(input.usageJson);
if (input.attributionMode) {
return {
attributionMode: input.attributionMode,
costCents:
typeof input.costCents === "number" && Number.isFinite(input.costCents)
? Math.max(0, Math.round(input.costCents))
: derivedCostCents ?? 0,
};
}
if (input.usageJson) {
return {
attributionMode: "included_in_run" as const,
costCents: derivedCostCents ?? 0,
};
}
return {
attributionMode: "untracked" as const,
costCents: 0,
};
}
function truncateSummary(summary: string | null): string | null {
if (!summary) {
return null;
}
return summary.length > 500 ? summary.slice(0, 500) : summary;
}
function deriveResultSummary(run: PostRunMemoryCaptureSource): string {
const commentSummary = truncateSummary(buildHeartbeatRunIssueComment(run.resultJson));
if (commentSummary) {
return commentSummary;
}
const summarizedResult = summarizeHeartbeatRunResultJson(run.resultJson);
const resultRecord = asRecord(summarizedResult);
const resultSummary = truncateSummary(
readNonEmptyString(resultRecord?.result)
?? readNonEmptyString(resultRecord?.message)
?? readNonEmptyString(resultRecord?.error)
?? readNonEmptyString(run.error),
);
if (resultSummary) {
return resultSummary;
}
return `Run ${run.id} finished with status ${run.status}`;
}
function buildQueuedCaptureResultJson(
run: PostRunMemoryCaptureSource,
resultJson: Record<string, unknown> | null,
): Record<string, unknown> {
const sourceResult = summarizeHeartbeatRunResultJson(run.resultJson);
return {
captureState: "queued",
hookKind: "post_run_capture",
sourceKind: "run",
sourceRunId: run.id,
sourceRunStatus: run.status,
...(run.errorCode ? { sourceErrorCode: run.errorCode } : {}),
...(run.sessionIdAfter ? { sourceSessionIdAfter: run.sessionIdAfter } : {}),
...(sourceResult ? { sourceResult } : {}),
...(resultJson ?? {}),
};
}
function mergeCaptureResultJson(
existing: Record<string, unknown> | null,
patch: Record<string, unknown>,
): Record<string, unknown> {
return {
...(existing ?? {}),
...patch,
};
}
export async function enqueuePostRunMemoryCaptureJob(
input: EnqueuePostRunMemoryCaptureInput,
) {
const bindingKey = input.bindingKey ?? DEFAULT_POST_RUN_MEMORY_BINDING_KEY;
const bindingId = input.bindingId ?? buildSyntheticMemoryBindingId(input.run.companyId, bindingKey);
const usageJson = input.usageJson ?? buildUsageSummaryFromRun(input.run.usageJson);
const attribution = resolveAttribution({
attributionMode: input.attributionMode,
costCents: input.costCents,
usageJson,
});
return input.store.enqueue({
companyId: input.run.companyId,
bindingId,
bindingKey,
operationType: "capture",
sourceKind: "run",
sourceAgentId: input.run.agentId,
sourceIssueId: input.sourceIssueId ?? null,
sourceProjectId: input.sourceProjectId ?? null,
sourceGoalId: input.sourceGoalId ?? null,
sourceHeartbeatRunId: input.run.id,
hookKind: "post_run_capture",
dispatcherKind: input.dispatcherKind ?? "in_process",
attributionMode: attribution.attributionMode,
costCents: attribution.costCents,
usageJson,
resultSummary: truncateSummary(input.resultSummary ?? deriveResultSummary(input.run)),
resultJson: buildQueuedCaptureResultJson(input.run, input.resultJson ?? null),
sourceRefJson: {
invocationSource: input.run.invocationSource,
...(input.run.triggerDetail ? { triggerDetail: input.run.triggerDetail } : {}),
sourceRunStatus: input.run.status,
...(input.run.sessionIdAfter ? { sessionIdAfter: input.run.sessionIdAfter } : {}),
...(input.sourceRefJson ?? {}),
},
});
}
export function createPostRunMemoryCaptureHandler(): MemoryJobHandler {
return async (job) => {
if (job.sourceKind !== "run" || job.hookKind !== "post_run_capture") {
return {
outcome: "failed",
errorCode: "unsupported_post_run_capture_source",
error: `Post-run capture handler only supports run/post_run_capture jobs; received ${job.sourceKind}/${job.hookKind ?? "null"}`,
resultSummary: job.resultSummary ?? "Unsupported post-run capture source",
resultJson: mergeCaptureResultJson(job.resultJson ?? null, { captureState: "failed" }),
};
}
if (!job.sourceHeartbeatRunId) {
return {
outcome: "failed",
errorCode: "source_run_missing",
error: "Post-run capture job is missing source heartbeat run provenance",
resultSummary: job.resultSummary ?? "Missing source run for post-run capture",
resultJson: mergeCaptureResultJson(job.resultJson ?? null, { captureState: "failed" }),
};
}
return {
outcome: "succeeded",
resultSummary: job.resultSummary ?? `Post-run capture recorded for run ${job.sourceHeartbeatRunId}`,
resultJson: mergeCaptureResultJson(job.resultJson ?? null, {
captureState: "succeeded",
sourceRunId: job.sourceHeartbeatRunId,
persistedUsageSummary: job.usageJson != null,
persistedSourceIssueId: job.sourceIssueId,
persistedSourceProjectId: job.sourceProjectId,
persistedSourceGoalId: job.sourceGoalId,
}),
};
};
}
@@ -0,0 +1,302 @@
import type {
MemoryExtractionJobAttributionMode,
MemoryExtractionJobUsage,
} from "@paperclipai/shared";
import { logger } from "../middleware/logger.js";
import type { MemoryJobStore } from "./memory-job-store.js";
const DEFAULT_POLL_INTERVAL_MS = 5_000;
const DEFAULT_RECOVERY_INTERVAL_MS = 30_000;
const DEFAULT_LEASE_DURATION_MS = 60_000;
const DEFAULT_MAX_CONCURRENT_JOBS = 2;
type ClaimedMemoryJob = NonNullable<Awaited<ReturnType<MemoryJobStore["claimNext"]>>>;
interface MemoryJobHandlerResultBase {
providerJobId?: string | null;
attributionMode?: MemoryExtractionJobAttributionMode;
costCents?: number;
resultSummary?: string | null;
usageJson?: MemoryExtractionJobUsage | null;
resultJson?: Record<string, unknown> | null;
}
export type MemoryJobHandlerResult =
| ({ outcome: "succeeded" } & MemoryJobHandlerResultBase)
| ({ outcome: "failed"; error?: string | null; errorCode?: string | null } & MemoryJobHandlerResultBase)
| ({ outcome: "running"; leaseDurationMs?: number | null } & MemoryJobHandlerResultBase);
export type MemoryJobHandler = (
job: ClaimedMemoryJob,
) => Promise<MemoryJobHandlerResult> | MemoryJobHandlerResult;
export interface MemoryJobDispatcherOptions {
store: MemoryJobStore;
resolveHandler: (job: ClaimedMemoryJob) => Promise<MemoryJobHandler | null> | MemoryJobHandler | null;
pollIntervalMs?: number;
recoveryIntervalMs?: number;
leaseDurationMs?: number;
maxConcurrentJobs?: number;
now?: () => Date;
}
export interface MemoryJobDispatcherDiagnostics {
running: boolean;
activeJobCount: number;
activeJobIds: string[];
tickCount: number;
lastTickAt: string | null;
recoveryCount: number;
lastRecoveryAt: string | null;
}
export interface MemoryJobDispatcher {
start(): void;
stop(): void;
tick(): Promise<void>;
sweepExpiredLeases(): Promise<Awaited<ReturnType<MemoryJobStore["recoverExpiredLeases"]>>>;
diagnostics(): MemoryJobDispatcherDiagnostics;
}
export function createMemoryJobDispatcher(
options: MemoryJobDispatcherOptions,
): MemoryJobDispatcher {
const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS;
const recoveryIntervalMs = options.recoveryIntervalMs ?? DEFAULT_RECOVERY_INTERVAL_MS;
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" });
let pollTimer: ReturnType<typeof setInterval> | null = null;
let recoveryTimer: ReturnType<typeof setInterval> | null = null;
let running = false;
let tickInProgress = false;
let recoveryInProgress = false;
let tickCount = 0;
let recoveryCount = 0;
let lastTickAt: Date | null = null;
let lastRecoveryAt: Date | null = null;
const activeJobs = new Set<string>();
async function runClaimedJob(job: ClaimedMemoryJob): Promise<void> {
const jobLog = log.child({
jobId: job.id,
companyId: job.companyId,
bindingKey: job.bindingKey,
operationType: job.operationType,
});
try {
const handler = await options.resolveHandler(job);
if (!handler) {
await options.store.fail({
companyId: job.companyId,
jobId: job.id,
now: now(),
errorCode: "handler_not_configured",
error: `No in-process memory job handler configured for ${job.operationType}`,
});
jobLog.warn("failed memory job because no handler was configured");
return;
}
const result = await handler(job);
const settledAt = now();
if (result.outcome === "succeeded") {
await options.store.complete({
companyId: job.companyId,
jobId: job.id,
now: settledAt,
...(result.providerJobId !== undefined ? { providerJobId: result.providerJobId } : {}),
...(result.attributionMode !== undefined ? { attributionMode: result.attributionMode } : {}),
...(result.costCents !== undefined ? { costCents: result.costCents } : {}),
...(result.resultSummary !== undefined ? { resultSummary: result.resultSummary } : {}),
...(result.usageJson !== undefined ? { usageJson: result.usageJson } : {}),
...(result.resultJson !== undefined ? { resultJson: result.resultJson } : {}),
});
jobLog.info("completed memory job successfully");
return;
}
if (result.outcome === "failed") {
await options.store.fail({
companyId: job.companyId,
jobId: job.id,
now: settledAt,
...(result.providerJobId !== undefined ? { providerJobId: result.providerJobId } : {}),
...(result.attributionMode !== undefined ? { attributionMode: result.attributionMode } : {}),
...(result.costCents !== undefined ? { costCents: result.costCents } : {}),
...(result.resultSummary !== undefined ? { resultSummary: result.resultSummary } : {}),
...(result.usageJson !== undefined ? { usageJson: result.usageJson } : {}),
...(result.resultJson !== undefined ? { resultJson: result.resultJson } : {}),
errorCode: result.errorCode,
error: result.error ?? "Memory job handler reported a failure",
});
jobLog.warn("memory job handler returned a failure result");
return;
}
await options.store.updateRunning({
companyId: job.companyId,
jobId: job.id,
now: settledAt,
...(result.providerJobId !== undefined ? { providerJobId: result.providerJobId } : {}),
...(result.attributionMode !== undefined ? { attributionMode: result.attributionMode } : {}),
...(result.costCents !== undefined ? { costCents: result.costCents } : {}),
...(result.resultSummary !== undefined ? { resultSummary: result.resultSummary } : {}),
...(result.usageJson !== undefined ? { usageJson: result.usageJson } : {}),
...(result.resultJson !== undefined ? { resultJson: result.resultJson } : {}),
leaseDurationMs: result.leaseDurationMs ?? leaseDurationMs,
});
jobLog.info("memory job remains running after provider handoff");
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
try {
await options.store.fail({
companyId: job.companyId,
jobId: job.id,
now: now(),
errorCode: "handler_exception",
error: message,
});
} catch (persistError) {
jobLog.error(
{ persistError: persistError instanceof Error ? persistError.message : String(persistError) },
"failed to persist handler exception on memory job",
);
}
jobLog.error({ error: message }, "memory job handler threw an exception");
} finally {
activeJobs.delete(job.id);
}
}
async function tick(): Promise<void> {
if (tickInProgress) {
return;
}
tickInProgress = true;
tickCount += 1;
lastTickAt = now();
const dispatches: Promise<void>[] = [];
try {
while (activeJobs.size < maxConcurrentJobs) {
const claimed = await options.store.claimNext({
dispatcherKind: "in_process",
leaseDurationMs,
now: now(),
});
if (!claimed) {
break;
}
activeJobs.add(claimed.id);
dispatches.push(runClaimedJob(claimed));
}
if (dispatches.length > 0) {
await Promise.allSettled(dispatches);
}
} catch (error) {
log.error(
{ error: error instanceof Error ? error.message : String(error) },
"memory job dispatcher tick failed",
);
} finally {
tickInProgress = false;
}
}
async function sweepExpiredLeases() {
if (recoveryInProgress) {
return { recovered: 0, jobIds: [] };
}
recoveryInProgress = true;
recoveryCount += 1;
lastRecoveryAt = now();
try {
const result = await options.store.recoverExpiredLeases({ now: now() });
if (result.recovered > 0) {
log.warn({ recovered: result.recovered, jobIds: result.jobIds }, "recovered expired memory job leases");
}
return result;
} catch (error) {
log.error(
{ error: error instanceof Error ? error.message : String(error) },
"memory job recovery sweep failed",
);
return { recovered: 0, jobIds: [] };
} finally {
recoveryInProgress = false;
}
}
function start(): void {
if (running) {
return;
}
running = true;
pollTimer = setInterval(() => {
void tick();
}, pollIntervalMs);
recoveryTimer = setInterval(() => {
void sweepExpiredLeases();
}, recoveryIntervalMs);
pollTimer.unref?.();
recoveryTimer.unref?.();
void sweepExpiredLeases();
void tick();
log.info({ pollIntervalMs, recoveryIntervalMs, leaseDurationMs, maxConcurrentJobs }, "memory job dispatcher started");
}
function stop(): void {
if (pollTimer !== null) {
clearInterval(pollTimer);
pollTimer = null;
}
if (recoveryTimer !== null) {
clearInterval(recoveryTimer);
recoveryTimer = null;
}
if (!running) {
return;
}
running = false;
log.info({ activeJobCount: activeJobs.size }, "memory job dispatcher stopped");
}
function diagnostics(): MemoryJobDispatcherDiagnostics {
return {
running,
activeJobCount: activeJobs.size,
activeJobIds: [...activeJobs],
tickCount,
lastTickAt: lastTickAt?.toISOString() ?? null,
recoveryCount,
lastRecoveryAt: lastRecoveryAt?.toISOString() ?? null,
};
}
return {
start,
stop,
tick,
sweepExpiredLeases,
diagnostics,
};
}
+757
View File
@@ -0,0 +1,757 @@
import {
and,
asc,
desc,
eq,
gte,
inArray,
isNull,
lte,
or,
sql,
} from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { memoryExtractionJobs } from "@paperclipai/db";
import type {
ListMemoryExtractionJobsQuery,
MemoryExtractionJob,
MemoryExtractionJobAttributionMode,
MemoryExtractionJobDetail,
MemoryExtractionJobDispatcherKind,
MemoryExtractionJobEffectiveState,
MemoryExtractionJobListResponse,
MemoryExtractionJobOperationType,
MemoryExtractionJobSourceKind,
MemoryExtractionJobSourceRef,
MemoryExtractionJobStatus,
MemoryExtractionJobUsage,
} from "@paperclipai/shared";
import { conflict, notFound } from "../errors.js";
type MemoryJobRecord = typeof memoryExtractionJobs.$inferSelect;
const DEFAULT_LIST_LIMIT = 50;
const DEFAULT_LIST_OFFSET = 0;
const CLAIM_RETRY_ATTEMPTS = 5;
const TERMINAL_MEMORY_JOB_STATUSES = new Set<MemoryExtractionJobStatus>([
"succeeded",
"failed",
"cancelled",
]);
export const DEFAULT_MEMORY_JOB_RECOVERY_ERROR_CODE = "lease_expired";
export const DEFAULT_MEMORY_JOB_RECOVERY_ERROR = "Memory job lease expired before completion";
export interface CreateMemoryJobInput {
companyId: string;
bindingId: string;
bindingKey: string;
operationType: MemoryExtractionJobOperationType;
sourceKind: MemoryExtractionJobSourceKind;
sourceAgentId?: string | null;
sourceIssueId?: string | null;
sourceProjectId?: string | null;
sourceGoalId?: string | null;
sourceHeartbeatRunId?: string | null;
hookKind?: MemoryJobRecord["hookKind"];
providerJobId?: string | null;
submittedAt?: Date;
attributionMode?: MemoryExtractionJobAttributionMode;
costCents?: number;
resultSummary?: string | null;
errorCode?: string | null;
error?: string | null;
sourceRefJson?: MemoryExtractionJobSourceRef | null;
dispatcherKind?: MemoryExtractionJobDispatcherKind;
usageJson?: MemoryExtractionJobUsage | null;
resultJson?: Record<string, unknown> | null;
}
export interface ClaimMemoryJobInput {
leaseDurationMs: number;
companyId?: string;
dispatcherKind?: MemoryExtractionJobDispatcherKind;
now?: Date;
}
interface MemoryJobLifecycleUpdate {
companyId: string;
jobId: string;
now?: Date;
providerJobId?: string | null;
attributionMode?: MemoryExtractionJobAttributionMode;
costCents?: number;
resultSummary?: string | null;
usageJson?: MemoryExtractionJobUsage | null;
resultJson?: Record<string, unknown> | null;
}
export interface UpdateRunningMemoryJobInput extends MemoryJobLifecycleUpdate {
leaseDurationMs?: number | null;
}
export interface CompleteMemoryJobInput extends MemoryJobLifecycleUpdate {}
export interface FailMemoryJobInput extends MemoryJobLifecycleUpdate {
errorCode?: string | null;
error?: string | null;
}
export interface CancelMemoryJobInput {
companyId: string;
jobId: string;
now?: Date;
resultSummary?: string | null;
errorCode?: string | null;
error?: string | null;
}
export interface RecoverExpiredMemoryJobsInput {
companyId?: string;
limit?: number;
now?: Date;
errorCode?: string;
error?: string;
}
export interface RecoverExpiredMemoryJobsResult {
recovered: number;
jobIds: string[];
}
function hasOwn(value: object, key: string): boolean {
return Object.prototype.hasOwnProperty.call(value, key);
}
function isMemoryJobStuck(
job: Pick<MemoryJobRecord, "status" | "leaseExpiresAt">,
now = new Date(),
): boolean {
return (
job.status === "running"
&& job.leaseExpiresAt !== null
&& job.leaseExpiresAt.getTime() <= now.getTime()
);
}
function effectiveStateForMemoryJob(
job: Pick<MemoryJobRecord, "status" | "leaseExpiresAt">,
now = new Date(),
): MemoryExtractionJobEffectiveState {
return isMemoryJobStuck(job, now) ? "stuck" : job.status;
}
function retryRootIdForJob(job: Pick<MemoryJobRecord, "id" | "retryOfJobId">): string {
return job.retryOfJobId ?? job.id;
}
function rerunEligibleForJob(job: Pick<MemoryJobRecord, "status">): boolean {
return TERMINAL_MEMORY_JOB_STATUSES.has(job.status);
}
function toMemoryExtractionJob(job: MemoryJobRecord): MemoryExtractionJob {
return {
id: job.id,
companyId: job.companyId,
bindingId: job.bindingId,
bindingKey: job.bindingKey,
operationType: job.operationType,
status: job.status,
sourceAgentId: job.sourceAgentId,
sourceIssueId: job.sourceIssueId,
sourceProjectId: job.sourceProjectId,
sourceGoalId: job.sourceGoalId,
sourceHeartbeatRunId: job.sourceHeartbeatRunId,
hookKind: job.hookKind,
providerJobId: job.providerJobId,
submittedAt: job.submittedAt,
startedAt: job.startedAt,
finishedAt: job.finishedAt,
attributionMode: job.attributionMode,
costCents: job.costCents,
resultSummary: job.resultSummary,
errorCode: job.errorCode,
error: job.error,
sourceKind: job.sourceKind,
sourceRefJson: (job.sourceRefJson as MemoryExtractionJobSourceRef | null) ?? null,
retryOfJobId: job.retryOfJobId,
attemptNumber: job.attemptNumber,
dispatcherKind: job.dispatcherKind,
leaseExpiresAt: job.leaseExpiresAt,
usageJson: (job.usageJson as MemoryExtractionJobUsage | null) ?? null,
resultJson: (job.resultJson as Record<string, unknown> | null) ?? null,
createdAt: job.createdAt,
updatedAt: job.updatedAt,
};
}
function withLifecycleUpdates(
base: Record<string, unknown>,
input: Partial<
MemoryJobLifecycleUpdate
& FailMemoryJobInput
& CancelMemoryJobInput
& UpdateRunningMemoryJobInput
>,
): Record<string, unknown> {
if (hasOwn(input, "providerJobId")) {
base.providerJobId = input.providerJobId ?? null;
}
if (hasOwn(input, "attributionMode")) {
base.attributionMode = input.attributionMode;
}
if (hasOwn(input, "costCents")) {
base.costCents = input.costCents;
}
if (hasOwn(input, "resultSummary")) {
base.resultSummary = input.resultSummary ?? null;
}
if (hasOwn(input, "usageJson")) {
base.usageJson = input.usageJson ?? null;
}
if (hasOwn(input, "resultJson")) {
base.resultJson = input.resultJson ?? null;
}
if (hasOwn(input, "errorCode")) {
base.errorCode = input.errorCode ?? null;
}
if (hasOwn(input, "error")) {
base.error = input.error ?? null;
}
return base;
}
async function ensureMemoryJobExists(
db: Db,
companyId: string,
jobId: string,
): Promise<MemoryJobRecord> {
const row = await db
.select()
.from(memoryExtractionJobs)
.where(and(eq(memoryExtractionJobs.companyId, companyId), eq(memoryExtractionJobs.id, jobId)))
.limit(1)
.then((rows) => rows[0] ?? null);
if (!row) {
throw notFound(`Memory job not found: ${jobId}`);
}
return row;
}
async function countRetryLineages(
db: Db,
companyId: string,
rootIds: string[],
): Promise<Map<string, number>> {
if (rootIds.length === 0) {
return new Map();
}
const uniqueRootIds = [...new Set(rootIds)];
const rows = await db
.select({
id: memoryExtractionJobs.id,
retryOfJobId: memoryExtractionJobs.retryOfJobId,
})
.from(memoryExtractionJobs)
.where(
and(
eq(memoryExtractionJobs.companyId, companyId),
or(
inArray(memoryExtractionJobs.id, uniqueRootIds),
inArray(memoryExtractionJobs.retryOfJobId, uniqueRootIds),
),
),
);
const counts = new Map<string, number>();
for (const row of rows) {
const rootId = row.retryOfJobId ?? row.id;
counts.set(rootId, (counts.get(rootId) ?? 0) + 1);
}
for (const rootId of uniqueRootIds) {
if (!counts.has(rootId)) {
counts.set(rootId, 1);
}
}
return counts;
}
export function memoryJobStore(db: Db) {
async function getJob(
companyId: string,
jobId: string,
): Promise<MemoryJobRecord | null> {
return db
.select()
.from(memoryExtractionJobs)
.where(and(eq(memoryExtractionJobs.companyId, companyId), eq(memoryExtractionJobs.id, jobId)))
.limit(1)
.then((rows) => rows[0] ?? null);
}
async function getDetail(
companyId: string,
jobId: string,
opts: { now?: Date } = {},
): Promise<MemoryExtractionJobDetail | null> {
const row = await getJob(companyId, jobId);
if (!row) {
return null;
}
const now = opts.now ?? new Date();
const rootId = retryRootIdForJob(row);
const retryCounts = await countRetryLineages(db, companyId, [rootId]);
const retryCount = Math.max((retryCounts.get(rootId) ?? 1) - 1, 0);
return {
...toMemoryExtractionJob(row),
effectiveState: effectiveStateForMemoryJob(row, now),
retryCount,
rerunEligible: rerunEligibleForJob(row),
};
}
async function list(
companyId: string,
query: Partial<ListMemoryExtractionJobsQuery> = {},
opts: { now?: Date } = {},
): Promise<MemoryExtractionJobListResponse> {
const now = opts.now ?? new Date();
const limit = query.limit ?? DEFAULT_LIST_LIMIT;
const offset = query.offset ?? DEFAULT_LIST_OFFSET;
const conditions = [eq(memoryExtractionJobs.companyId, companyId)];
if (query.status) {
conditions.push(eq(memoryExtractionJobs.status, query.status));
}
if (query.effectiveState === "stuck") {
conditions.push(eq(memoryExtractionJobs.status, "running"));
conditions.push(lte(memoryExtractionJobs.leaseExpiresAt, now));
} else if (query.effectiveState === "running") {
conditions.push(eq(memoryExtractionJobs.status, "running"));
conditions.push(sql`(${memoryExtractionJobs.leaseExpiresAt} is null or ${memoryExtractionJobs.leaseExpiresAt} > ${now})`);
} else if (query.effectiveState) {
conditions.push(eq(memoryExtractionJobs.status, query.effectiveState));
}
if (query.bindingKey) {
conditions.push(eq(memoryExtractionJobs.bindingKey, query.bindingKey));
}
if (query.operationType) {
conditions.push(eq(memoryExtractionJobs.operationType, query.operationType));
}
if (query.agentId) {
conditions.push(eq(memoryExtractionJobs.sourceAgentId, query.agentId));
}
if (query.issueId) {
conditions.push(eq(memoryExtractionJobs.sourceIssueId, query.issueId));
}
if (query.runId) {
conditions.push(eq(memoryExtractionJobs.sourceHeartbeatRunId, query.runId));
}
if (query.submittedAfter) {
conditions.push(gte(memoryExtractionJobs.submittedAt, query.submittedAfter));
}
if (query.submittedBefore) {
conditions.push(lte(memoryExtractionJobs.submittedAt, query.submittedBefore));
}
const rows = await db
.select()
.from(memoryExtractionJobs)
.where(and(...conditions))
.orderBy(desc(memoryExtractionJobs.submittedAt), desc(memoryExtractionJobs.createdAt))
.limit(limit + 1)
.offset(offset);
const visibleRows = rows.slice(0, limit);
const retryCounts = await countRetryLineages(
db,
companyId,
visibleRows.map((row) => retryRootIdForJob(row)),
);
return {
jobs: visibleRows.map((row) => {
const rootId = retryRootIdForJob(row);
return {
id: row.id,
bindingId: row.bindingId,
bindingKey: row.bindingKey,
operationType: row.operationType,
status: row.status,
effectiveState: effectiveStateForMemoryJob(row, now),
sourceAgentId: row.sourceAgentId,
sourceIssueId: row.sourceIssueId,
sourceProjectId: row.sourceProjectId,
sourceGoalId: row.sourceGoalId,
sourceHeartbeatRunId: row.sourceHeartbeatRunId,
hookKind: row.hookKind,
providerJobId: row.providerJobId,
submittedAt: row.submittedAt,
startedAt: row.startedAt,
finishedAt: row.finishedAt,
attributionMode: row.attributionMode,
costCents: row.costCents,
resultSummary: row.resultSummary,
errorCode: row.errorCode,
error: row.error,
retryOfJobId: row.retryOfJobId,
attemptNumber: row.attemptNumber,
retryCount: Math.max((retryCounts.get(rootId) ?? 1) - 1, 0),
};
}),
nextOffset: rows.length > limit ? offset + limit : null,
};
}
async function enqueue(input: CreateMemoryJobInput): Promise<MemoryJobRecord> {
return db
.insert(memoryExtractionJobs)
.values({
companyId: input.companyId,
bindingId: input.bindingId,
bindingKey: input.bindingKey,
operationType: input.operationType,
sourceKind: input.sourceKind,
sourceAgentId: input.sourceAgentId ?? null,
sourceIssueId: input.sourceIssueId ?? null,
sourceProjectId: input.sourceProjectId ?? null,
sourceGoalId: input.sourceGoalId ?? null,
sourceHeartbeatRunId: input.sourceHeartbeatRunId ?? null,
hookKind: input.hookKind ?? null,
providerJobId: input.providerJobId ?? null,
submittedAt: input.submittedAt,
attributionMode: input.attributionMode ?? "untracked",
costCents: input.costCents ?? 0,
resultSummary: input.resultSummary ?? null,
errorCode: input.errorCode ?? null,
error: input.error ?? null,
sourceRefJson: (input.sourceRefJson as Record<string, unknown> | null | undefined) ?? null,
dispatcherKind: input.dispatcherKind ?? "in_process",
usageJson: (input.usageJson as Record<string, unknown> | null | undefined) ?? null,
resultJson: input.resultJson ?? null,
})
.returning()
.then((rows) => rows[0]);
}
async function claimNext(input: ClaimMemoryJobInput): Promise<MemoryJobRecord | null> {
const now = input.now ?? new Date();
const dispatcherKind = input.dispatcherKind ?? "in_process";
const leaseExpiresAt = new Date(now.getTime() + input.leaseDurationMs);
for (let attempt = 0; attempt < CLAIM_RETRY_ATTEMPTS; attempt++) {
const conditions = [
eq(memoryExtractionJobs.status, "queued"),
eq(memoryExtractionJobs.dispatcherKind, dispatcherKind),
];
if (input.companyId) {
conditions.push(eq(memoryExtractionJobs.companyId, input.companyId));
}
const candidate = await db
.select()
.from(memoryExtractionJobs)
.where(and(...conditions))
.orderBy(asc(memoryExtractionJobs.submittedAt), asc(memoryExtractionJobs.createdAt))
.limit(1)
.then((rows) => rows[0] ?? null);
if (!candidate) {
return null;
}
const claimed = await db
.update(memoryExtractionJobs)
.set({
status: "running",
startedAt: candidate.startedAt ?? now,
leaseExpiresAt,
updatedAt: now,
})
.where(
and(
eq(memoryExtractionJobs.id, candidate.id),
eq(memoryExtractionJobs.companyId, candidate.companyId),
eq(memoryExtractionJobs.status, "queued"),
),
)
.returning()
.then((rows) => rows[0] ?? null);
if (claimed) {
return claimed;
}
}
return null;
}
async function updateRunning(input: UpdateRunningMemoryJobInput): Promise<MemoryJobRecord> {
const now = input.now ?? new Date();
const updates = withLifecycleUpdates({ updatedAt: now }, input);
if (hasOwn(input, "leaseDurationMs")) {
updates.leaseExpiresAt = input.leaseDurationMs === null
? null
: new Date(now.getTime() + (input.leaseDurationMs ?? 0));
}
const updated = await db
.update(memoryExtractionJobs)
.set(updates)
.where(
and(
eq(memoryExtractionJobs.companyId, input.companyId),
eq(memoryExtractionJobs.id, input.jobId),
eq(memoryExtractionJobs.status, "running"),
),
)
.returning()
.then((rows) => rows[0] ?? null);
if (updated) {
return updated;
}
const existing = await ensureMemoryJobExists(db, input.companyId, input.jobId);
throw conflict(`Memory job ${existing.id} is not running`);
}
async function complete(input: CompleteMemoryJobInput): Promise<MemoryJobRecord> {
const now = input.now ?? new Date();
const updated = await db
.update(memoryExtractionJobs)
.set(
withLifecycleUpdates(
{
status: "succeeded",
finishedAt: now,
leaseExpiresAt: null,
errorCode: null,
error: null,
updatedAt: now,
},
input,
),
)
.where(
and(
eq(memoryExtractionJobs.companyId, input.companyId),
eq(memoryExtractionJobs.id, input.jobId),
eq(memoryExtractionJobs.status, "running"),
),
)
.returning()
.then((rows) => rows[0] ?? null);
if (updated) {
return updated;
}
const existing = await ensureMemoryJobExists(db, input.companyId, input.jobId);
throw conflict(`Memory job ${existing.id} is not running`);
}
async function fail(input: FailMemoryJobInput): Promise<MemoryJobRecord> {
const now = input.now ?? new Date();
const updated = await db
.update(memoryExtractionJobs)
.set(
withLifecycleUpdates(
{
status: "failed",
finishedAt: now,
leaseExpiresAt: null,
updatedAt: now,
},
input,
),
)
.where(
and(
eq(memoryExtractionJobs.companyId, input.companyId),
eq(memoryExtractionJobs.id, input.jobId),
eq(memoryExtractionJobs.status, "running"),
),
)
.returning()
.then((rows) => rows[0] ?? null);
if (updated) {
return updated;
}
const existing = await ensureMemoryJobExists(db, input.companyId, input.jobId);
throw conflict(`Memory job ${existing.id} is not running`);
}
async function cancel(input: CancelMemoryJobInput): Promise<MemoryJobRecord> {
const now = input.now ?? new Date();
const updated = await db
.update(memoryExtractionJobs)
.set(
withLifecycleUpdates(
{
status: "cancelled",
finishedAt: now,
leaseExpiresAt: null,
updatedAt: now,
},
input,
),
)
.where(
and(
eq(memoryExtractionJobs.companyId, input.companyId),
eq(memoryExtractionJobs.id, input.jobId),
inArray(memoryExtractionJobs.status, ["queued", "running"]),
),
)
.returning()
.then((rows) => rows[0] ?? null);
if (updated) {
return updated;
}
const existing = await ensureMemoryJobExists(db, input.companyId, input.jobId);
throw conflict(`Memory job ${existing.id} cannot be cancelled from status ${existing.status}`);
}
async function recoverExpiredLeases(
input: RecoverExpiredMemoryJobsInput = {},
): Promise<RecoverExpiredMemoryJobsResult> {
const now = input.now ?? new Date();
const limit = input.limit ?? 100;
const conditions = [eq(memoryExtractionJobs.status, "running"), lte(memoryExtractionJobs.leaseExpiresAt, now)];
if (input.companyId) {
conditions.push(eq(memoryExtractionJobs.companyId, input.companyId));
}
const expiredRows = await db
.select({ id: memoryExtractionJobs.id })
.from(memoryExtractionJobs)
.where(and(...conditions))
.orderBy(asc(memoryExtractionJobs.leaseExpiresAt), asc(memoryExtractionJobs.createdAt))
.limit(limit);
const expiredIds = expiredRows.map((row) => row.id);
if (expiredIds.length === 0) {
return { recovered: 0, jobIds: [] };
}
const updated = await db
.update(memoryExtractionJobs)
.set({
status: "failed",
finishedAt: now,
leaseExpiresAt: null,
updatedAt: now,
errorCode: input.errorCode ?? DEFAULT_MEMORY_JOB_RECOVERY_ERROR_CODE,
error: input.error ?? DEFAULT_MEMORY_JOB_RECOVERY_ERROR,
})
.where(
and(
inArray(memoryExtractionJobs.id, expiredIds),
eq(memoryExtractionJobs.status, "running"),
),
)
.returning({ id: memoryExtractionJobs.id });
return {
recovered: updated.length,
jobIds: updated.map((row) => row.id),
};
}
async function rerun(
companyId: string,
jobId: string,
opts: { now?: Date } = {},
): Promise<MemoryJobRecord> {
const sourceJob = await ensureMemoryJobExists(db, companyId, jobId);
if (!rerunEligibleForJob(sourceJob)) {
throw conflict(`Memory job ${sourceJob.id} cannot be rerun from status ${sourceJob.status}`);
}
const retryRootId = retryRootIdForJob(sourceJob);
const lineageRows = await db
.select({ attemptNumber: memoryExtractionJobs.attemptNumber })
.from(memoryExtractionJobs)
.where(
and(
eq(memoryExtractionJobs.companyId, companyId),
or(
eq(memoryExtractionJobs.id, retryRootId),
eq(memoryExtractionJobs.retryOfJobId, retryRootId),
),
),
);
const nextAttemptNumber = lineageRows.reduce((maxAttempt, row) => {
return Math.max(maxAttempt, row.attemptNumber);
}, 0) + 1;
return db
.insert(memoryExtractionJobs)
.values({
companyId: sourceJob.companyId,
bindingId: sourceJob.bindingId,
bindingKey: sourceJob.bindingKey,
operationType: sourceJob.operationType,
status: "queued",
sourceAgentId: sourceJob.sourceAgentId,
sourceIssueId: sourceJob.sourceIssueId,
sourceProjectId: sourceJob.sourceProjectId,
sourceGoalId: sourceJob.sourceGoalId,
sourceHeartbeatRunId: sourceJob.sourceHeartbeatRunId,
hookKind: sourceJob.hookKind,
submittedAt: opts.now,
attributionMode: sourceJob.attributionMode,
costCents: 0,
resultSummary: null,
errorCode: null,
error: null,
sourceKind: sourceJob.sourceKind,
sourceRefJson: sourceJob.sourceRefJson,
retryOfJobId: retryRootId,
attemptNumber: nextAttemptNumber,
dispatcherKind: sourceJob.dispatcherKind,
leaseExpiresAt: null,
providerJobId: null,
usageJson: null,
resultJson: null,
startedAt: null,
finishedAt: null,
})
.returning()
.then((rows) => rows[0]);
}
return {
getJob,
getDetail,
list,
enqueue,
claimNext,
updateRunning,
complete,
fail,
cancel,
recoverExpiredLeases,
rerun,
};
}
export type MemoryJobStore = ReturnType<typeof memoryJobStore>;
export { effectiveStateForMemoryJob, isMemoryJobStuck };
+2
View File
@@ -25,6 +25,7 @@ import { Costs } from "./pages/Costs";
import { Activity } from "./pages/Activity";
import { Inbox } from "./pages/Inbox";
import { CompanySettings } from "./pages/CompanySettings";
import { MemoryJobs } from "./pages/MemoryJobs";
import { CompanyAccess } from "./pages/CompanyAccess";
import { CompanyInvites } from "./pages/CompanyInvites";
import { CompanySkills } from "./pages/CompanySkills";
@@ -61,6 +62,7 @@ function boardRoutes() {
<Route path="onboarding" element={<OnboardingRoutePage />} />
<Route path="companies" element={<Companies />} />
<Route path="company/settings" element={<CompanySettings />} />
<Route path="company/settings/memory-jobs" element={<MemoryJobs />} />
<Route path="company/settings/access" element={<CompanyAccess />} />
<Route path="company/settings/invites" element={<CompanyInvites />} />
<Route path="company/export/*" element={<CompanyExport />} />
+1
View File
@@ -11,6 +11,7 @@ export { goalsApi } from "./goals";
export { approvalsApi } from "./approvals";
export { costsApi } from "./costs";
export { activityApi } from "./activity";
export { memoryJobsApi } from "./memoryJobs";
export { dashboardApi } from "./dashboard";
export { heartbeatsApi } from "./heartbeats";
export { instanceSettingsApi } from "./instanceSettings";
+47
View File
@@ -0,0 +1,47 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
const mockApi = vi.hoisted(() => ({
get: vi.fn(),
post: vi.fn(),
}));
vi.mock("./client", () => ({
api: mockApi,
}));
import { memoryJobsApi } from "./memoryJobs";
describe("memoryJobsApi", () => {
beforeEach(() => {
mockApi.get.mockReset();
mockApi.post.mockReset();
mockApi.get.mockResolvedValue({ jobs: [], nextOffset: null });
mockApi.post.mockResolvedValue({ job: { id: "job-1" } });
});
it("serializes list filters into the company-scoped memory jobs path", async () => {
await memoryJobsApi.list("company-1", {
effectiveState: "stuck",
bindingKey: "primary",
operationType: "capture",
agentId: "agent-1",
issueId: "issue-1",
runId: "run-1",
limit: 25,
offset: 50,
});
expect(mockApi.get).toHaveBeenCalledWith(
"/companies/company-1/memory/jobs?effectiveState=stuck&bindingKey=primary&operationType=capture&agentId=agent-1&issueId=issue-1&runId=run-1&limit=25&offset=50",
);
});
it("posts reruns to the selected job route", async () => {
await memoryJobsApi.rerun("company-1", "job-1");
expect(mockApi.post).toHaveBeenCalledWith(
"/companies/company-1/memory/jobs/job-1/rerun",
{},
);
});
});
+54
View File
@@ -0,0 +1,54 @@
import type {
MemoryExtractionJobDetailResponse,
MemoryExtractionJobEffectiveState,
MemoryExtractionJobListResponse,
MemoryExtractionJobOperationType,
MemoryExtractionJobRerunResponse,
MemoryExtractionJobStatus,
} from "@paperclipai/shared";
import { api } from "./client";
export interface MemoryJobListFilters {
status?: MemoryExtractionJobStatus;
effectiveState?: MemoryExtractionJobEffectiveState;
bindingKey?: string;
operationType?: MemoryExtractionJobOperationType;
agentId?: string;
issueId?: string;
runId?: string;
submittedAfter?: Date | string;
submittedBefore?: Date | string;
limit?: number;
offset?: number;
}
function setDateParam(params: URLSearchParams, key: string, value: Date | string | undefined) {
if (!value) return;
params.set(key, value instanceof Date ? value.toISOString() : value);
}
export const memoryJobsApi = {
list: (companyId: string, filters: MemoryJobListFilters = {}) => {
const params = new URLSearchParams();
if (filters.status) params.set("status", filters.status);
if (filters.effectiveState) params.set("effectiveState", filters.effectiveState);
if (filters.bindingKey) params.set("bindingKey", filters.bindingKey);
if (filters.operationType) params.set("operationType", filters.operationType);
if (filters.agentId) params.set("agentId", filters.agentId);
if (filters.issueId) params.set("issueId", filters.issueId);
if (filters.runId) params.set("runId", filters.runId);
setDateParam(params, "submittedAfter", filters.submittedAfter);
setDateParam(params, "submittedBefore", filters.submittedBefore);
if (filters.limit != null) params.set("limit", String(filters.limit));
if (filters.offset != null) params.set("offset", String(filters.offset));
const qs = params.toString();
return api.get<MemoryExtractionJobListResponse>(
`/companies/${companyId}/memory/jobs${qs ? `?${qs}` : ""}`,
);
},
get: (companyId: string, jobId: string) =>
api.get<MemoryExtractionJobDetailResponse>(`/companies/${companyId}/memory/jobs/${jobId}`),
rerun: (companyId: string, jobId: string) =>
api.post<MemoryExtractionJobRerunResponse>(`/companies/${companyId}/memory/jobs/${jobId}/rerun`, {}),
};
+2 -1
View File
@@ -1,5 +1,5 @@
import { useQuery } from "@tanstack/react-query";
import { ChevronLeft, MailPlus, Settings, Shield, SlidersHorizontal } from "lucide-react";
import { ChevronLeft, Database, MailPlus, Settings, Shield, SlidersHorizontal } from "lucide-react";
import { sidebarBadgesApi } from "@/api/sidebarBadges";
import { ApiError } from "@/api/client";
import { Link } from "@/lib/router";
@@ -54,6 +54,7 @@ export function CompanySettingsSidebar() {
<nav className="flex-1 min-h-0 overflow-y-auto scrollbar-auto-hide px-3 py-2">
<div className="flex flex-col gap-0.5">
<SidebarNavItem to="/company/settings" label="General" icon={SlidersHorizontal} end />
<SidebarNavItem to="/company/settings/memory-jobs" label="Memory Jobs" icon={Database} end />
<SidebarNavItem
to="/company/settings/access"
label="Access"
+6
View File
@@ -90,6 +90,12 @@ export const queryKeys = {
comments: (approvalId: string) => ["approvals", "comments", approvalId] as const,
issues: (approvalId: string) => ["approvals", "issues", approvalId] as const,
},
memoryJobs: {
list: (companyId: string, filters?: Record<string, unknown>) =>
["memory-jobs", companyId, filters ?? {}] as const,
detail: (companyId: string, jobId: string) =>
["memory-jobs", companyId, "detail", jobId] as const,
},
access: {
invites: (companyId: string, state: string = "all", limit: number = 20) =>
["access", "invites", "paginated-v1", companyId, state, limit] as const,
+800
View File
@@ -0,0 +1,800 @@
import { useEffect, useMemo, useState } from "react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import {
isUuidLike,
MEMORY_EXTRACTION_JOB_OPERATION_TYPES,
MEMORY_EXTRACTION_JOB_STATUSES,
type Agent,
type HeartbeatRun,
type Issue,
type MemoryExtractionJobDetail,
type MemoryExtractionJobEffectiveState,
type MemoryExtractionJobListItem,
type MemoryExtractionJobOperationType,
} from "@paperclipai/shared";
import { agentsApi } from "@/api/agents";
import { ApiError } from "@/api/client";
import { heartbeatsApi } from "@/api/heartbeats";
import { issuesApi } from "@/api/issues";
import { memoryJobsApi } from "@/api/memoryJobs";
import { useBreadcrumbs } from "@/context/BreadcrumbContext";
import { useCompany } from "@/context/CompanyContext";
import { useToastActions } from "@/context/ToastContext";
import { Link, useSearchParams } from "@/lib/router";
import { cn } from "@/lib/utils";
import { queryKeys } from "@/lib/queryKeys";
import { EmptyState } from "@/components/EmptyState";
import { PageSkeleton } from "@/components/PageSkeleton";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
import { Input } from "@/components/ui/input";
import {
Select,
SelectContent,
SelectItem,
SelectTrigger,
SelectValue,
} from "@/components/ui/select";
import { AlertTriangle, Database, RefreshCcw, Search } from "lucide-react";
const JOB_LIST_LIMIT = 100;
const BINDING_KEY_PATTERN = /^[a-z0-9][a-z0-9_-]*$/;
type JobStateFilter = "attention" | "all" | MemoryExtractionJobEffectiveState;
type FilterDraft = {
state: JobStateFilter;
bindingKey: string;
operationType: "all" | MemoryExtractionJobOperationType;
agentId: string;
issueRef: string;
runId: string;
};
type AppliedFilters = {
state: JobStateFilter;
bindingKey?: string;
operationType?: MemoryExtractionJobOperationType;
agentId?: string;
issueId?: string;
issueLabel?: string;
runId?: string;
};
const defaultDraftFilters: FilterDraft = {
state: "attention",
bindingKey: "",
operationType: "all",
agentId: "all",
issueRef: "",
runId: "",
};
const defaultAppliedFilters: AppliedFilters = {
state: "attention",
};
function formatLabel(value: string) {
return value.replaceAll("_", " ");
}
function formatTimestamp(value: Date | string | null | undefined) {
if (!value) return "—";
return new Date(value).toLocaleString();
}
function formatCost(cents: number | null | undefined) {
if (typeof cents !== "number") return "—";
return new Intl.NumberFormat(undefined, {
style: "currency",
currency: "USD",
minimumFractionDigits: 2,
}).format(cents / 100);
}
function shortId(value: string | null | undefined) {
if (!value) return null;
return value.length > 12 ? `${value.slice(0, 8)}…` : value;
}
function summarizeJob(job: Pick<MemoryExtractionJobListItem | MemoryExtractionJobDetail, "error" | "resultSummary">) {
if (job.error) return job.error;
if (job.resultSummary) return job.resultSummary;
return "—";
}
function mergeJobLists(...jobs: MemoryExtractionJobListItem[][]) {
return jobs
.flat()
.sort((left, right) => {
return new Date(right.submittedAt).getTime() - new Date(left.submittedAt).getTime();
});
}
function stateBadgeTone(value: MemoryExtractionJobEffectiveState) {
switch (value) {
case "failed":
return "destructive" as const;
case "stuck":
return "secondary" as const;
case "succeeded":
return "secondary" as const;
case "queued":
case "cancelled":
return "outline" as const;
default:
return "default" as const;
}
}
function JsonBlock({ value }: { value: unknown }) {
if (value == null) {
return <span className="text-sm text-muted-foreground">None</span>;
}
return (
<pre className="max-h-80 overflow-auto rounded-md border border-border bg-muted/30 p-3 text-xs leading-5">
{JSON.stringify(value, null, 2)}
</pre>
);
}
function DetailValue({ label, value }: { label: string; value: React.ReactNode }) {
return (
<div className="space-y-1">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">{label}</div>
<div className="text-sm text-foreground">{value}</div>
</div>
);
}
export function MemoryJobs() {
const { selectedCompanyId } = useCompany();
const { setBreadcrumbs } = useBreadcrumbs();
const { pushToast } = useToastActions();
const queryClient = useQueryClient();
const [searchParams, setSearchParams] = useSearchParams();
const [draftFilters, setDraftFilters] = useState<FilterDraft>(defaultDraftFilters);
const [appliedFilters, setAppliedFilters] = useState<AppliedFilters>(defaultAppliedFilters);
const [filterError, setFilterError] = useState<string | null>(null);
const [isApplyingFilters, setIsApplyingFilters] = useState(false);
useEffect(() => {
setBreadcrumbs([
{ label: "Company Settings", href: "/company/settings" },
{ label: "Memory Jobs" },
]);
}, [setBreadcrumbs]);
const { data: agents } = useQuery({
queryKey: selectedCompanyId ? queryKeys.agents.list(selectedCompanyId) : ["agents", "__none__"],
queryFn: () => agentsApi.list(selectedCompanyId!),
enabled: !!selectedCompanyId,
});
const agentById = useMemo(() => {
const map = new Map<string, Agent>();
for (const agent of agents ?? []) {
map.set(agent.id, agent);
}
return map;
}, [agents]);
const commonFilters = useMemo(
() => ({
bindingKey: appliedFilters.bindingKey,
operationType: appliedFilters.operationType,
agentId: appliedFilters.agentId,
issueId: appliedFilters.issueId,
runId: appliedFilters.runId,
limit: JOB_LIST_LIMIT,
}),
[
appliedFilters.agentId,
appliedFilters.bindingKey,
appliedFilters.issueId,
appliedFilters.operationType,
appliedFilters.runId,
],
);
const singleStateFilter: MemoryExtractionJobEffectiveState | undefined =
appliedFilters.state === "attention" || appliedFilters.state === "all"
? undefined
: appliedFilters.state;
const attentionFailedQuery = useQuery({
queryKey: selectedCompanyId
? queryKeys.memoryJobs.list(selectedCompanyId, { ...commonFilters, effectiveState: "failed" })
: ["memory-jobs", "__none__", "failed"],
queryFn: () => memoryJobsApi.list(selectedCompanyId!, { ...commonFilters, effectiveState: "failed" }),
enabled: !!selectedCompanyId && appliedFilters.state === "attention",
});
const attentionStuckQuery = useQuery({
queryKey: selectedCompanyId
? queryKeys.memoryJobs.list(selectedCompanyId, { ...commonFilters, effectiveState: "stuck" })
: ["memory-jobs", "__none__", "stuck"],
queryFn: () => memoryJobsApi.list(selectedCompanyId!, { ...commonFilters, effectiveState: "stuck" }),
enabled: !!selectedCompanyId && appliedFilters.state === "attention",
});
const listQuery = useQuery({
queryKey: selectedCompanyId
? queryKeys.memoryJobs.list(selectedCompanyId, {
...commonFilters,
effectiveState: singleStateFilter,
})
: ["memory-jobs", "__none__", "list"],
queryFn: () =>
memoryJobsApi.list(selectedCompanyId!, {
...commonFilters,
effectiveState: singleStateFilter,
}),
enabled: !!selectedCompanyId && appliedFilters.state !== "attention",
});
const jobs = useMemo(() => {
if (appliedFilters.state === "attention") {
return mergeJobLists(attentionFailedQuery.data?.jobs ?? [], attentionStuckQuery.data?.jobs ?? []);
}
return listQuery.data?.jobs ?? [];
}, [
appliedFilters.state,
attentionFailedQuery.data?.jobs,
attentionStuckQuery.data?.jobs,
listQuery.data?.jobs,
]);
const listError = (appliedFilters.state === "attention"
? attentionFailedQuery.error ?? attentionStuckQuery.error
: listQuery.error) as Error | null;
const listLoading = appliedFilters.state === "attention"
? attentionFailedQuery.isLoading || attentionStuckQuery.isLoading
: listQuery.isLoading;
const explicitSelectedJobId = searchParams.get("jobId");
const selectedJobId = explicitSelectedJobId ?? jobs[0]?.id ?? null;
const detailQuery = useQuery({
queryKey: selectedCompanyId && selectedJobId
? queryKeys.memoryJobs.detail(selectedCompanyId, selectedJobId)
: ["memory-jobs", "detail", "__none__"],
queryFn: () => memoryJobsApi.get(selectedCompanyId!, selectedJobId!),
enabled: !!selectedCompanyId && !!selectedJobId,
});
const selectedJob = detailQuery.data?.job ?? null;
const sourceIssueQuery = useQuery({
queryKey: selectedJob?.sourceIssueId ? queryKeys.issues.detail(selectedJob.sourceIssueId) : ["issues", "detail", "__none__"],
queryFn: () => issuesApi.get(selectedJob!.sourceIssueId!),
enabled: !!selectedJob?.sourceIssueId,
retry: false,
});
const sourceRunQuery = useQuery({
queryKey: selectedJob?.sourceHeartbeatRunId
? queryKeys.runDetail(selectedJob.sourceHeartbeatRunId)
: ["heartbeat-run", "__none__"],
queryFn: () => heartbeatsApi.get(selectedJob!.sourceHeartbeatRunId!),
enabled: !!selectedJob?.sourceHeartbeatRunId,
retry: false,
});
const rerunMutation = useMutation({
mutationFn: (jobId: string) => memoryJobsApi.rerun(selectedCompanyId!, jobId),
onSuccess: (result) => {
queryClient.setQueryData(queryKeys.memoryJobs.detail(selectedCompanyId!, result.job.id), result);
queryClient.invalidateQueries({ queryKey: queryKeys.memoryJobs.list(selectedCompanyId!) });
queryClient.invalidateQueries({ queryKey: queryKeys.memoryJobs.detail(selectedCompanyId!, result.job.id) });
const shouldSwitchToAll = appliedFilters.state !== "all";
if (shouldSwitchToAll) {
setAppliedFilters((current) => ({ ...current, state: "all" }));
setDraftFilters((current) => ({ ...current, state: "all" }));
}
setSearchParams((current) => {
const next = new URLSearchParams(current);
next.set("jobId", result.job.id);
return next;
});
pushToast({
title: "Memory job rerun queued",
body: shouldSwitchToAll ? "Switched the list to All so the new queued attempt stays visible." : undefined,
tone: "success",
});
},
onError: (error) => {
pushToast({
title: "Memory job rerun failed",
body: error instanceof Error ? error.message : "Unknown error",
tone: "error",
});
},
});
async function applyFilters() {
setIsApplyingFilters(true);
setFilterError(null);
try {
const nextBindingKey = draftFilters.bindingKey.trim();
if (nextBindingKey && !BINDING_KEY_PATTERN.test(nextBindingKey)) {
setFilterError("Binding key must use lowercase letters, numbers, _ or -.");
return;
}
const nextRunId = draftFilters.runId.trim();
if (nextRunId && !isUuidLike(nextRunId)) {
setFilterError("Run filter expects a heartbeat run UUID.");
return;
}
let issueId: string | undefined;
let issueLabel: string | undefined;
const nextIssueRef = draftFilters.issueRef.trim();
if (nextIssueRef) {
if (isUuidLike(nextIssueRef)) {
issueId = nextIssueRef;
issueLabel = nextIssueRef;
} else {
const issue = await issuesApi.get(nextIssueRef);
if (issue.companyId !== selectedCompanyId) {
setFilterError("Issue filter resolved outside the selected company.");
return;
}
issueId = issue.id;
issueLabel = issue.identifier ?? issue.id;
}
}
setAppliedFilters({
state: draftFilters.state,
bindingKey: nextBindingKey || undefined,
operationType: draftFilters.operationType === "all" ? undefined : draftFilters.operationType,
agentId: draftFilters.agentId === "all" ? undefined : draftFilters.agentId,
issueId,
issueLabel,
runId: nextRunId || undefined,
});
setSearchParams((current) => {
const next = new URLSearchParams(current);
next.delete("jobId");
return next;
});
} catch (error) {
if (error instanceof ApiError && error.status === 404) {
setFilterError("Issue filter did not match a visible issue.");
return;
}
setFilterError(error instanceof Error ? error.message : "Failed to apply filters.");
} finally {
setIsApplyingFilters(false);
}
}
function resetFilters() {
setFilterError(null);
setDraftFilters(defaultDraftFilters);
setAppliedFilters(defaultAppliedFilters);
setSearchParams((current) => {
const next = new URLSearchParams(current);
next.delete("jobId");
return next;
});
}
const selectedIssue = sourceIssueQuery.data as Issue | undefined;
const selectedRun = sourceRunQuery.data as HeartbeatRun | undefined;
if (!selectedCompanyId) {
return <EmptyState icon={Database} message="Select a company to view memory jobs." />;
}
if (listLoading && jobs.length === 0) {
return <PageSkeleton variant="detail" />;
}
return (
<div className="space-y-4">
<Card>
<CardHeader>
<CardTitle>Memory Jobs</CardTitle>
<CardDescription>
Inspect failed and stuck memory extraction work, trace provenance, and rerun eligible attempts.
</CardDescription>
</CardHeader>
<CardContent className="space-y-4">
<div className="grid gap-3 md:grid-cols-2 xl:grid-cols-6">
<div className="space-y-1 xl:col-span-1">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">State</div>
<Select
value={draftFilters.state}
onValueChange={(value) => setDraftFilters((current) => ({ ...current, state: value as JobStateFilter }))}
>
<SelectTrigger>
<SelectValue placeholder="All states" />
</SelectTrigger>
<SelectContent>
<SelectItem value="attention">Failed + stuck</SelectItem>
<SelectItem value="all">All states</SelectItem>
<SelectItem value="stuck">Stuck</SelectItem>
{MEMORY_EXTRACTION_JOB_STATUSES.map((status) => (
<SelectItem key={status} value={status}>
{formatLabel(status)}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
<div className="space-y-1 xl:col-span-1">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Operation</div>
<Select
value={draftFilters.operationType}
onValueChange={(value) =>
setDraftFilters((current) => ({
...current,
operationType: value as FilterDraft["operationType"],
}))
}
>
<SelectTrigger>
<SelectValue placeholder="All operations" />
</SelectTrigger>
<SelectContent>
<SelectItem value="all">All operations</SelectItem>
{MEMORY_EXTRACTION_JOB_OPERATION_TYPES.map((operationType) => (
<SelectItem key={operationType} value={operationType}>
{formatLabel(operationType)}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
<div className="space-y-1 xl:col-span-1">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Agent</div>
<Select
value={draftFilters.agentId}
onValueChange={(value) => setDraftFilters((current) => ({ ...current, agentId: value }))}
>
<SelectTrigger>
<SelectValue placeholder="All agents" />
</SelectTrigger>
<SelectContent>
<SelectItem value="all">All agents</SelectItem>
{(agents ?? []).map((agent) => (
<SelectItem key={agent.id} value={agent.id}>
{agent.name}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
<div className="space-y-1 xl:col-span-1">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Binding key</div>
<Input
value={draftFilters.bindingKey}
onChange={(event) => setDraftFilters((current) => ({ ...current, bindingKey: event.target.value }))}
placeholder="primary"
/>
</div>
<div className="space-y-1 xl:col-span-1">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Issue</div>
<Input
value={draftFilters.issueRef}
onChange={(event) => setDraftFilters((current) => ({ ...current, issueRef: event.target.value }))}
placeholder="PAP-1708 or UUID"
/>
</div>
<div className="space-y-1 xl:col-span-1">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Run</div>
<Input
value={draftFilters.runId}
onChange={(event) => setDraftFilters((current) => ({ ...current, runId: event.target.value }))}
placeholder="Heartbeat run UUID"
/>
</div>
</div>
<div className="flex flex-col gap-3 md:flex-row md:items-center md:justify-between">
<div className="flex flex-wrap items-center gap-2 text-xs text-muted-foreground">
<Badge variant="outline">Showing latest {JOB_LIST_LIMIT}</Badge>
<Badge variant="outline">{jobs.length} visible</Badge>
{appliedFilters.issueLabel ? <Badge variant="outline">Issue {appliedFilters.issueLabel}</Badge> : null}
</div>
<div className="flex gap-2">
<Button variant="outline" onClick={resetFilters}>Reset</Button>
<Button onClick={applyFilters} disabled={isApplyingFilters}>
<Search className="h-4 w-4" />
Apply filters
</Button>
</div>
</div>
{filterError ? <p className="text-sm text-destructive">{filterError}</p> : null}
{listError ? <p className="text-sm text-destructive">{listError.message}</p> : null}
</CardContent>
</Card>
<div className="grid gap-4 xl:grid-cols-[minmax(0,1.2fr)_minmax(360px,0.8fr)]">
<Card className="overflow-hidden py-0">
<CardHeader className="border-b py-4">
<CardTitle className="text-base">Job List</CardTitle>
<CardDescription>
{appliedFilters.state === "attention"
? "Failed and stuck jobs across the selected filters."
: "Newest jobs matching the selected filters."}
</CardDescription>
</CardHeader>
<CardContent className="px-0">
{jobs.length === 0 ? (
<EmptyState icon={Database} message="No memory jobs match these filters." />
) : (
<div className="overflow-auto">
<table className="w-full min-w-[860px] text-sm">
<thead className="border-b bg-muted/30 text-xs uppercase tracking-wide text-muted-foreground">
<tr>
<th className="px-4 py-3 text-left font-medium">Job</th>
<th className="px-4 py-3 text-left font-medium">State</th>
<th className="px-4 py-3 text-left font-medium">Binding</th>
<th className="px-4 py-3 text-left font-medium">Issue</th>
<th className="px-4 py-3 text-left font-medium">Run</th>
<th className="px-4 py-3 text-left font-medium">Agent</th>
<th className="px-4 py-3 text-left font-medium">Submitted</th>
<th className="px-4 py-3 text-left font-medium">Cost</th>
<th className="px-4 py-3 text-left font-medium">Summary</th>
</tr>
</thead>
<tbody>
{jobs.map((job) => {
const isSelected = selectedJobId === job.id;
return (
<tr key={job.id} className={cn("border-b align-top", isSelected && "bg-accent/30")}>
<td className="px-4 py-3">
<button
type="button"
className="space-y-1 text-left"
onClick={() => {
setSearchParams((current) => {
const next = new URLSearchParams(current);
next.set("jobId", job.id);
return next;
});
}}
>
<div className="font-medium text-foreground">{formatLabel(job.operationType)}</div>
<div className="text-xs text-muted-foreground">{shortId(job.id)}</div>
</button>
</td>
<td className="px-4 py-3">
<div className="flex flex-wrap gap-2">
<Badge variant={stateBadgeTone(job.effectiveState)}>{formatLabel(job.effectiveState)}</Badge>
<Badge variant="outline">Attempt {job.attemptNumber}</Badge>
</div>
</td>
<td className="px-4 py-3 text-muted-foreground">{job.bindingKey}</td>
<td className="px-4 py-3">
{job.sourceIssueId ? (
<Link className="text-primary hover:underline" to={`/issues/${job.sourceIssueId}`}>
{shortId(job.sourceIssueId)}
</Link>
) : (
<span className="text-muted-foreground">—</span>
)}
</td>
<td className="px-4 py-3">
{job.sourceHeartbeatRunId && job.sourceAgentId ? (
<Link className="text-primary hover:underline" to={`/agents/${job.sourceAgentId}/runs/${job.sourceHeartbeatRunId}`}>
{shortId(job.sourceHeartbeatRunId)}
</Link>
) : job.sourceHeartbeatRunId ? (
<span className="text-muted-foreground">{shortId(job.sourceHeartbeatRunId)}</span>
) : (
<span className="text-muted-foreground">—</span>
)}
</td>
<td className="px-4 py-3">
{job.sourceAgentId ? (
<Link className="text-primary hover:underline" to={`/agents/${job.sourceAgentId}`}>
{agentById.get(job.sourceAgentId)?.name ?? shortId(job.sourceAgentId)}
</Link>
) : (
<span className="text-muted-foreground">—</span>
)}
</td>
<td className="px-4 py-3 text-muted-foreground">{formatTimestamp(job.submittedAt)}</td>
<td className="px-4 py-3 text-muted-foreground">{formatCost(job.costCents)}</td>
<td className="max-w-xs px-4 py-3">
<div className={cn("line-clamp-3 text-sm", job.error ? "text-destructive" : "text-muted-foreground")}>
{summarizeJob(job)}
</div>
</td>
</tr>
);
})}
</tbody>
</table>
</div>
)}
</CardContent>
</Card>
<Card>
<CardHeader>
<CardTitle className="flex items-center gap-2 text-base">
Job Detail
{selectedJob ? <Badge variant="outline">{shortId(selectedJob.id)}</Badge> : null}
</CardTitle>
<CardDescription>
Provenance, lifecycle, usage, and rerun controls for the selected memory job.
</CardDescription>
</CardHeader>
<CardContent className="space-y-5">
{selectedJobId && detailQuery.isLoading ? (
<PageSkeleton variant="detail" />
) : detailQuery.error ? (
<div className="rounded-md border border-destructive/40 bg-destructive/5 p-4 text-sm text-destructive">
{(detailQuery.error as Error).message}
</div>
) : selectedJob ? (
<>
<div className="flex flex-wrap items-center gap-2">
<Badge variant={stateBadgeTone(selectedJob.effectiveState)}>
{formatLabel(selectedJob.effectiveState)}
</Badge>
<Badge variant="outline">{formatLabel(selectedJob.status)}</Badge>
<Badge variant="outline">Attempt {selectedJob.attemptNumber}</Badge>
{selectedJob.retryCount > 0 ? (
<Badge variant="outline">{selectedJob.retryCount} prior retries</Badge>
) : null}
</div>
{selectedJob.error ? (
<div className="rounded-md border border-amber-500/30 bg-amber-500/10 p-4 text-sm text-amber-950 dark:text-amber-200">
<div className="mb-1 flex items-center gap-2 font-medium">
<AlertTriangle className="h-4 w-4" />
Failure surfaced to operators
</div>
<div>{selectedJob.error}</div>
</div>
) : null}
<div className="grid gap-4 md:grid-cols-2">
<DetailValue label="Operation" value={formatLabel(selectedJob.operationType)} />
<DetailValue label="Binding key" value={selectedJob.bindingKey} />
<DetailValue label="Hook kind" value={selectedJob.hookKind ? formatLabel(selectedJob.hookKind) : "—"} />
<DetailValue label="Source kind" value={formatLabel(selectedJob.sourceKind)} />
<DetailValue label="Dispatcher" value={formatLabel(selectedJob.dispatcherKind)} />
<DetailValue label="Attribution" value={formatLabel(selectedJob.attributionMode)} />
<DetailValue label="Submitted" value={formatTimestamp(selectedJob.submittedAt)} />
<DetailValue label="Started" value={formatTimestamp(selectedJob.startedAt)} />
<DetailValue label="Finished" value={formatTimestamp(selectedJob.finishedAt)} />
<DetailValue label="Lease expires" value={formatTimestamp(selectedJob.leaseExpiresAt)} />
<DetailValue label="Cost" value={formatCost(selectedJob.costCents)} />
<DetailValue label="Provider job id" value={selectedJob.providerJobId ?? "—"} />
</div>
<div className="space-y-3">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Provenance</div>
<div className="grid gap-4 md:grid-cols-2">
<DetailValue
label="Issue"
value={selectedJob.sourceIssueId ? (
<Link className="text-primary hover:underline" to={`/issues/${selectedJob.sourceIssueId}`}>
{selectedIssue?.identifier ?? shortId(selectedJob.sourceIssueId)}
</Link>
) : "—"}
/>
<DetailValue
label="Run"
value={selectedJob.sourceHeartbeatRunId && selectedJob.sourceAgentId ? (
<Link
className="text-primary hover:underline"
to={`/agents/${selectedJob.sourceAgentId}/runs/${selectedJob.sourceHeartbeatRunId}`}
>
{selectedRun?.id ? shortId(selectedRun.id) : shortId(selectedJob.sourceHeartbeatRunId)}
</Link>
) : selectedJob.sourceHeartbeatRunId ? shortId(selectedJob.sourceHeartbeatRunId) : "—"}
/>
<DetailValue
label="Agent"
value={selectedJob.sourceAgentId ? (
<Link className="text-primary hover:underline" to={`/agents/${selectedJob.sourceAgentId}`}>
{agentById.get(selectedJob.sourceAgentId)?.name ?? shortId(selectedJob.sourceAgentId)}
</Link>
) : "—"}
/>
<DetailValue
label="Retry root"
value={selectedJob.retryOfJobId ? shortId(selectedJob.retryOfJobId) : "Original attempt"}
/>
<DetailValue
label="Project"
value={selectedJob.sourceProjectId ? (
<Link className="text-primary hover:underline" to={`/projects/${selectedJob.sourceProjectId}`}>
{shortId(selectedJob.sourceProjectId)}
</Link>
) : "—"}
/>
<DetailValue
label="Goal"
value={selectedJob.sourceGoalId ? (
<Link className="text-primary hover:underline" to={`/goals/${selectedJob.sourceGoalId}`}>
{shortId(selectedJob.sourceGoalId)}
</Link>
) : "—"}
/>
</div>
</div>
<div className="space-y-3">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Summary</div>
<div className="rounded-md border border-border bg-muted/20 p-3 text-sm">
{summarizeJob(selectedJob)}
</div>
{selectedJob.errorCode ? (
<div className="text-sm text-muted-foreground">Error code: <span className="font-medium text-foreground">{selectedJob.errorCode}</span></div>
) : null}
</div>
<div className="space-y-3">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Usage</div>
<div className="grid gap-4 md:grid-cols-2">
<DetailValue label="Provider" value={selectedJob.usageJson?.provider ?? "—"} />
<DetailValue label="Model" value={selectedJob.usageJson?.model ?? "—"} />
<DetailValue label="Biller" value={selectedJob.usageJson?.biller ?? "—"} />
<DetailValue label="Billing type" value={selectedJob.usageJson?.billingType ?? "—"} />
<DetailValue label="Input tokens" value={selectedJob.usageJson?.inputTokens ?? "—"} />
<DetailValue label="Cached input" value={selectedJob.usageJson?.cachedInputTokens ?? "—"} />
<DetailValue label="Output tokens" value={selectedJob.usageJson?.outputTokens ?? "—"} />
<DetailValue label="Embedding tokens" value={selectedJob.usageJson?.embeddingTokens ?? "—"} />
<DetailValue label="Latency" value={selectedJob.usageJson?.latencyMs ? `${selectedJob.usageJson.latencyMs} ms` : "—"} />
<DetailValue label="Usage cost" value={typeof selectedJob.usageJson?.costCents === "number" ? formatCost(selectedJob.usageJson.costCents) : "—"} />
</div>
<JsonBlock value={selectedJob.usageJson?.details ?? null} />
</div>
<div className="space-y-3">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Source ref</div>
<JsonBlock value={selectedJob.sourceRefJson} />
</div>
<div className="space-y-3">
<div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Result payload</div>
<JsonBlock value={selectedJob.resultJson} />
</div>
<div className="flex flex-wrap gap-2 border-t pt-4">
<Button
onClick={() => rerunMutation.mutate(selectedJob.id)}
disabled={!selectedJob.rerunEligible || rerunMutation.isPending}
>
<RefreshCcw className="h-4 w-4" />
{rerunMutation.isPending ? "Queuing rerun..." : "Rerun job"}
</Button>
{!selectedJob.rerunEligible ? (
<span className="text-sm text-muted-foreground">
Rerun is available after the job reaches a terminal state.
</span>
) : null}
</div>
</>
) : (
<EmptyState icon={Database} message="Select a memory job to inspect details." />
)}
</CardContent>
</Card>
</div>
</div>
);
}