Skip to content

Add durable workflow orchestration foundation #78

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Waishnav wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Add durable workflow orchestration foundation #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2b18101
feat(workflows): add durable workflow schema
Waishnav Jul 16, 2026
1134de8
feat(workflows): add durable orchestration service
Waishnav Jul 16, 2026
fa9cef5
test(workflows): cover durable orchestration
Waishnav Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@
"dev": "node scripts/dev-server.mjs",
"postinstall": "node scripts/fix-node-pty-permissions.mjs",
"start": "node dist/cli.js serve",
"test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts",
"test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && npm run test:workflows && tsx src/cli.test.ts",
"test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/workflows.test.ts",
"typecheck": "tsc -p tsconfig.json --noEmit"
},
"keywords": [],
Expand Down
90 changes: 90 additions & 0 deletions src/db/migrations.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,11 @@ const migrations: Migration[] = [
name: "local-agent-sessions",
up: migrateLocalAgentSessions,
},
{
version: 4,
name: "durable-workflows",
up: migrateDurableWorkflows,
},
];

export function migrateDatabase(sqlite: Database.Database): void {
Expand Down Expand Up @@ -174,6 +179,91 @@ function migrateLocalAgentSessions(sqlite: Database.Database): void {
addColumnIfMissing(sqlite, "local_agent_sessions", "thinking", "text");
}

function migrateDurableWorkflows(sqlite: Database.Database): void {
sqlite.exec(`
create table if not exists workflow_runs (
id text primary key,
definition_version integer not null,
status text not null,
definition_json text not null,
input_json text not null,
policy_json text not null,
idempotency_key text unique,
request_hash text not null,
result_json text,
error_json text,
cancellation_requested_at text,
event_sequence integer not null default 0,
created_at text not null,
updated_at text not null,
started_at text,
completed_at text
);

create index if not exists workflow_runs_status_idx
on workflow_runs(status, created_at);

create table if not exists workflow_nodes (
id text primary key,
workflow_run_id text not null,
node_key text not null,
node_type text not null,
status text not null,
definition_json text not null,
attempt integer not null default 0,
claim_token text,
claimed_at text,
claim_expires_at text,
result_json text,
error_json text,
created_at text not null,
updated_at text not null,
completed_at text,
foreign key (workflow_run_id) references workflow_runs(id) on delete cascade
);

create unique index if not exists workflow_nodes_run_key_idx
on workflow_nodes(workflow_run_id, node_key);

create unique index if not exists workflow_nodes_run_id_idx
on workflow_nodes(workflow_run_id, id);

create index if not exists workflow_nodes_status_idx
on workflow_nodes(workflow_run_id, status, created_at);

create table if not exists workflow_edges (
workflow_run_id text not null,
from_node_id text not null,
to_node_id text not null,
primary key (workflow_run_id, from_node_id, to_node_id),
foreign key (workflow_run_id) references workflow_runs(id) on delete cascade,
foreign key (workflow_run_id, from_node_id)
references workflow_nodes(workflow_run_id, id) on delete cascade,
foreign key (workflow_run_id, to_node_id)
references workflow_nodes(workflow_run_id, id) on delete cascade
);

create index if not exists workflow_edges_to_node_idx
on workflow_edges(workflow_run_id, to_node_id);

create table if not exists workflow_events (
workflow_run_id text not null,
sequence integer not null,
event_type text not null,
node_id text,
payload_json text not null,
created_at text not null,
primary key (workflow_run_id, sequence),
foreign key (workflow_run_id) references workflow_runs(id) on delete cascade,
foreign key (workflow_run_id, node_id)
references workflow_nodes(workflow_run_id, id)
);

create index if not exists workflow_events_cursor_idx
on workflow_events(workflow_run_id, sequence);
`);
Comment on lines +184 to +264

@coderabbitai coderabbitai Bot Jul 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Fail migration 4 when workflow objects already exist.

IF NOT EXISTS can silently accept an incompatible legacy table that happens to contain the indexed columns, record version 4, and leave runtime queries failing on missing columns or constraints.

  • src/db/migrations.ts#L184-L264: use plain CREATE TABLE/CREATE INDEX, or explicitly verify every existing object's columns, keys, and foreign keys before recording the migration.
  • src/workflows/migrations.test.ts#L172-L208: cover a legacy workflow_events table containing workflow_run_id and sequence but missing required payload/event columns; migration must still fail and roll back.
📍 Affects 2 files
  • src/db/migrations.ts#L184-L264 (this comment)
  • src/workflows/migrations.test.ts#L172-L208
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/db/migrations.ts` around lines 184 - 264, Make migration 4 fail rather
than silently accepting incompatible existing workflow objects: in the workflow
schema creation block of src/db/migrations.ts (lines 184-264), replace the IF
NOT EXISTS table/index creation or validate every existing object’s columns,
keys, and foreign keys before recording the migration. In
src/workflows/migrations.test.ts (lines 172-208), add coverage for a legacy
workflow_events table containing workflow_run_id and sequence but missing
required event/payload columns, and assert the migration fails and rolls back.

}

function addColumnIfMissing(
sqlite: Database.Database,
table: "workspace_sessions" | "local_agent_sessions",
Expand Down
114 changes: 113 additions & 1 deletion src/db/schema.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,12 @@
import { index, integer, primaryKey, sqliteTable, text } from "drizzle-orm/sqlite-core";
import {
foreignKey,
index,
integer,
primaryKey,
sqliteTable,
text,
uniqueIndex,
} from "drizzle-orm/sqlite-core";

export const workspaceSessions = sqliteTable(
"workspace_sessions",
Expand Down Expand Up @@ -97,9 +105,113 @@ export const localAgentSessions = sqliteTable(
],
);

export const workflowRuns = sqliteTable(
"workflow_runs",
{
id: text("id").primaryKey(),
definitionVersion: integer("definition_version").notNull(),
status: text("status").notNull(),
definitionJson: text("definition_json").notNull(),
inputJson: text("input_json").notNull(),
policyJson: text("policy_json").notNull(),
idempotencyKey: text("idempotency_key").unique(),
requestHash: text("request_hash").notNull(),
resultJson: text("result_json"),
errorJson: text("error_json"),
cancellationRequestedAt: text("cancellation_requested_at"),
eventSequence: integer("event_sequence").notNull().default(0),
createdAt: text("created_at").notNull(),
updatedAt: text("updated_at").notNull(),
startedAt: text("started_at"),
completedAt: text("completed_at"),
},
(table) => [index("workflow_runs_status_idx").on(table.status, table.createdAt)],
);

export const workflowNodes = sqliteTable(
"workflow_nodes",
{
id: text("id").primaryKey(),
workflowRunId: text("workflow_run_id")
.notNull()
.references(() => workflowRuns.id, { onDelete: "cascade" }),
nodeKey: text("node_key").notNull(),
nodeType: text("node_type").notNull(),
status: text("status").notNull(),
definitionJson: text("definition_json").notNull(),
attempt: integer("attempt").notNull().default(0),
claimToken: text("claim_token"),
claimedAt: text("claimed_at"),
claimExpiresAt: text("claim_expires_at"),
resultJson: text("result_json"),
errorJson: text("error_json"),
createdAt: text("created_at").notNull(),
updatedAt: text("updated_at").notNull(),
completedAt: text("completed_at"),
},
(table) => [
uniqueIndex("workflow_nodes_run_key_idx").on(table.workflowRunId, table.nodeKey),
uniqueIndex("workflow_nodes_run_id_idx").on(table.workflowRunId, table.id),
index("workflow_nodes_status_idx").on(table.workflowRunId, table.status, table.createdAt),
],
);

export const workflowEdges = sqliteTable(
"workflow_edges",
{
workflowRunId: text("workflow_run_id")
.notNull()
.references(() => workflowRuns.id, { onDelete: "cascade" }),
fromNodeId: text("from_node_id").notNull(),
toNodeId: text("to_node_id").notNull(),
},
(table) => [
primaryKey({ columns: [table.workflowRunId, table.fromNodeId, table.toNodeId] }),
foreignKey({
columns: [table.workflowRunId, table.fromNodeId],
foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id],
}).onDelete("cascade"),
foreignKey({
columns: [table.workflowRunId, table.toNodeId],
foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id],
}).onDelete("cascade"),
index("workflow_edges_to_node_idx").on(table.workflowRunId, table.toNodeId),
],
);

export const workflowEvents = sqliteTable(
"workflow_events",
{
workflowRunId: text("workflow_run_id")
.notNull()
.references(() => workflowRuns.id, { onDelete: "cascade" }),
sequence: integer("sequence").notNull(),
eventType: text("event_type").notNull(),
nodeId: text("node_id"),
payloadJson: text("payload_json").notNull(),
createdAt: text("created_at").notNull(),
},
(table) => [
primaryKey({ columns: [table.workflowRunId, table.sequence] }),
foreignKey({
columns: [table.workflowRunId, table.nodeId],
foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id],
}),
index("workflow_events_cursor_idx").on(table.workflowRunId, table.sequence),
],
);

export type WorkspaceSessionRow = typeof workspaceSessions.$inferSelect;
export type NewWorkspaceSessionRow = typeof workspaceSessions.$inferInsert;
export type LoadedAgentFileRow = typeof loadedAgentFiles.$inferSelect;
export type NewLoadedAgentFileRow = typeof loadedAgentFiles.$inferInsert;
export type LocalAgentSessionRow = typeof localAgentSessions.$inferSelect;
export type NewLocalAgentSessionRow = typeof localAgentSessions.$inferInsert;
export type WorkflowRunRow = typeof workflowRuns.$inferSelect;
export type NewWorkflowRunRow = typeof workflowRuns.$inferInsert;
export type WorkflowNodeRow = typeof workflowNodes.$inferSelect;
export type NewWorkflowNodeRow = typeof workflowNodes.$inferInsert;
export type WorkflowEdgeRow = typeof workflowEdges.$inferSelect;
export type NewWorkflowEdgeRow = typeof workflowEdges.$inferInsert;
export type WorkflowEventRow = typeof workflowEvents.$inferSelect;
export type NewWorkflowEventRow = typeof workflowEvents.$inferInsert;
1 change: 1 addition & 0 deletions src/oauth-store.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise<void> {
{ version: 1, name: "workspace-state" },
{ version: 2, name: "oauth-state" },
{ version: 3, name: "local-agent-sessions" },
{ version: 4, name: "durable-workflows" },
]);
} finally {
database.close();
Expand Down
Loading
Loading