Skip to content

Commit 04f4a91

Browse files
badcubanbadcuban
authored
Sync inbox lifecycle state across devices (#94)
Keep Active/Wrapped filing and unread state synchronized across desktop and relay-connected devices.
1 parent ba0b8d1 commit 04f4a91

52 files changed

Lines changed: 1507 additions & 197 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,8 @@ describe("OrchestrationEngine", () => {
150150
updatedAt: "2026-03-03T00:00:03.000Z",
151151
archivedAt: null,
152152
pinnedAt: null,
153+
doneOverride: null,
154+
lastSeenAt: null,
153155
deletedAt: null,
154156
messages: [],
155157
proposedPlans: [],
@@ -485,6 +487,85 @@ describe("OrchestrationEngine", () => {
485487
await system.dispose();
486488
});
487489

490+
it("keeps inbox lifecycle state on the thread so every device reads the same answer", async () => {
491+
const system = await createOrchestrationSystem();
492+
const { engine } = system;
493+
const createdAt = now();
494+
495+
await system.run(
496+
engine.dispatch({
497+
type: "project.create",
498+
commandId: CommandId.make("cmd-project-inbox-create"),
499+
projectId: asProjectId("project-inbox"),
500+
title: "Project Inbox",
501+
workspaceRoot: "/tmp/project-inbox",
502+
defaultModelSelection: {
503+
instanceId: ProviderInstanceId.make("codex"),
504+
model: "gpt-5-codex",
505+
},
506+
createdAt,
507+
}),
508+
);
509+
await system.run(
510+
engine.dispatch({
511+
type: "thread.create",
512+
commandId: CommandId.make("cmd-thread-inbox-create"),
513+
threadId: ThreadId.make("thread-inbox"),
514+
projectId: asProjectId("project-inbox"),
515+
title: "File me",
516+
modelSelection: {
517+
instanceId: ProviderInstanceId.make("codex"),
518+
model: "gpt-5-codex",
519+
},
520+
interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
521+
runtimeMode: "full-access",
522+
branch: null,
523+
worktreePath: null,
524+
createdAt,
525+
}),
526+
);
527+
const readThread = async () =>
528+
(await system.readModel()).threads.find((thread) => thread.id === "thread-inbox");
529+
const updatedAtAfterCreate = (await readThread())?.updatedAt;
530+
531+
await system.run(
532+
engine.dispatch({
533+
type: "thread.done-override.set",
534+
commandId: CommandId.make("cmd-thread-inbox-done"),
535+
threadId: ThreadId.make("thread-inbox"),
536+
state: "done",
537+
at: "2026-01-02T09:30:00.000Z",
538+
}),
539+
);
540+
await system.run(
541+
engine.dispatch({
542+
type: "thread.seen.set",
543+
commandId: CommandId.make("cmd-thread-inbox-seen"),
544+
threadId: ThreadId.make("thread-inbox"),
545+
at: "2026-01-02T09:31:00.000Z",
546+
}),
547+
);
548+
// Mark unread walks the stamp back to just before the completion it
549+
// un-sees, so a later write with an earlier stamp has to win.
550+
await system.run(
551+
engine.dispatch({
552+
type: "thread.seen.set",
553+
commandId: CommandId.make("cmd-thread-inbox-unread"),
554+
threadId: ThreadId.make("thread-inbox"),
555+
at: "2026-01-02T09:29:59.999Z",
556+
}),
557+
);
558+
559+
const thread = await readThread();
560+
expect(thread?.doneOverride).toEqual({ state: "done", at: "2026-01-02T09:30:00.000Z" });
561+
expect(thread?.lastSeenAt).toBe("2026-01-02T09:29:59.999Z");
562+
// Filing and reading are not work: `updatedAt` is what the inbox weighs
563+
// these stamps against, and what auto-wrap idles off.
564+
expect(thread?.updatedAt).toBe(updatedAtAfterCreate);
565+
566+
await system.dispose();
567+
});
568+
488569
it("replays append-only events from sequence", async () => {
489570
const system = await createOrchestrationSystem();
490571
const { engine } = system;

‎apps/server/src/orchestration/Layers/ProjectionPipeline.ts‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -565,6 +565,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
565565
updatedAt: event.payload.updatedAt,
566566
archivedAt: null,
567567
pinnedAt: null,
568+
doneOverride: null,
569+
doneOverrideAt: null,
570+
lastSeenAt: null,
568571
latestUserMessageAt: null,
569572
pendingApprovalCount: 0,
570573
pendingUserInputCount: 0,
@@ -633,6 +636,37 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
633636
return;
634637
}
635638

639+
// Inbox filing and read state deliberately leave `updatedAt` alone --
640+
// the client weighs both stamps against the thread's real activity.
641+
case "thread.done-override-set": {
642+
const existingRow = yield* projectionThreadRepository.getById({
643+
threadId: event.payload.threadId,
644+
});
645+
if (Option.isNone(existingRow)) {
646+
return;
647+
}
648+
yield* projectionThreadRepository.upsert({
649+
...existingRow.value,
650+
doneOverride: event.payload.state,
651+
doneOverrideAt: event.payload.at,
652+
});
653+
return;
654+
}
655+
656+
case "thread.seen-set": {
657+
const existingRow = yield* projectionThreadRepository.getById({
658+
threadId: event.payload.threadId,
659+
});
660+
if (Option.isNone(existingRow)) {
661+
return;
662+
}
663+
yield* projectionThreadRepository.upsert({
664+
...existingRow.value,
665+
lastSeenAt: event.payload.at,
666+
});
667+
return;
668+
}
669+
636670
case "thread.meta-updated": {
637671
const existingRow = yield* projectionThreadRepository.getById({
638672
threadId: event.payload.threadId,

‎apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -315,6 +315,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
315315
updatedAt: "2026-02-24T00:00:03.000Z",
316316
archivedAt: null,
317317
pinnedAt: null,
318+
doneOverride: null,
319+
lastSeenAt: null,
318320
deletedAt: null,
319321
messages: [
320322
{
@@ -435,6 +437,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
435437
updatedAt: "2026-02-24T00:00:03.000Z",
436438
archivedAt: null,
437439
pinnedAt: null,
440+
doneOverride: null,
441+
lastSeenAt: null,
438442
session: {
439443
threadId: ThreadId.make("thread-1"),
440444
status: "running",

‎apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import {
2121
type OrchestrationSession,
2222
type OrchestrationThreadActivity,
2323
type OrchestrationThreadDiffStat,
24+
type OrchestrationThreadDoneOverride,
2425
type OrchestrationThreadShell,
2526
ModelSelection,
2627
OrchestrationThreadGoal,
@@ -231,6 +232,15 @@ function mapLatestTurn(
231232
};
232233
}
233234

235+
function mapThreadDoneOverride(
236+
row: Schema.Schema.Type<typeof ProjectionThreadDbRowSchema>,
237+
): OrchestrationThreadDoneOverride | null {
238+
if (row.doneOverride == null || row.doneOverrideAt == null) {
239+
return null;
240+
}
241+
return { state: row.doneOverride, at: row.doneOverrideAt };
242+
}
243+
234244
function mapThreadDiffStat(
235245
row: Schema.Schema.Type<typeof ProjectionThreadDiffStatDbRowSchema> | undefined,
236246
): OrchestrationThreadDiffStat | null {
@@ -392,6 +402,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
392402
updated_at AS "updatedAt",
393403
archived_at AS "archivedAt",
394404
pinned_at AS "pinnedAt",
405+
done_override AS "doneOverride",
406+
done_override_at AS "doneOverrideAt",
407+
last_seen_at AS "lastSeenAt",
395408
latest_user_message_at AS "latestUserMessageAt",
396409
pending_approval_count AS "pendingApprovalCount",
397410
pending_user_input_count AS "pendingUserInputCount",
@@ -425,6 +438,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
425438
updated_at AS "updatedAt",
426439
archived_at AS "archivedAt",
427440
pinned_at AS "pinnedAt",
441+
done_override AS "doneOverride",
442+
done_override_at AS "doneOverrideAt",
443+
last_seen_at AS "lastSeenAt",
428444
latest_user_message_at AS "latestUserMessageAt",
429445
pending_approval_count AS "pendingApprovalCount",
430446
pending_user_input_count AS "pendingUserInputCount",
@@ -460,6 +476,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
460476
updated_at AS "updatedAt",
461477
archived_at AS "archivedAt",
462478
pinned_at AS "pinnedAt",
479+
done_override AS "doneOverride",
480+
done_override_at AS "doneOverrideAt",
481+
last_seen_at AS "lastSeenAt",
463482
latest_user_message_at AS "latestUserMessageAt",
464483
pending_approval_count AS "pendingApprovalCount",
465484
pending_user_input_count AS "pendingUserInputCount",
@@ -977,6 +996,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
977996
updated_at AS "updatedAt",
978997
archived_at AS "archivedAt",
979998
pinned_at AS "pinnedAt",
999+
done_override AS "doneOverride",
1000+
done_override_at AS "doneOverrideAt",
1001+
last_seen_at AS "lastSeenAt",
9801002
latest_user_message_at AS "latestUserMessageAt",
9811003
pending_approval_count AS "pendingApprovalCount",
9821004
pending_user_input_count AS "pendingUserInputCount",
@@ -1423,6 +1445,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
14231445
updatedAt: row.updatedAt,
14241446
archivedAt: row.archivedAt,
14251447
pinnedAt: row.pinnedAt,
1448+
doneOverride: mapThreadDoneOverride(row),
1449+
lastSeenAt: row.lastSeenAt ?? null,
14261450
deletedAt: row.deletedAt,
14271451
messages: messagesByThread.get(row.threadId) ?? [],
14281452
proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
@@ -1660,6 +1684,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
16601684
updatedAt: row.updatedAt,
16611685
archivedAt: row.archivedAt,
16621686
pinnedAt: row.pinnedAt,
1687+
doneOverride: mapThreadDoneOverride(row),
1688+
lastSeenAt: row.lastSeenAt ?? null,
16631689
deletedAt: row.deletedAt,
16641690
messages: messagesByThread.get(row.threadId) ?? [],
16651691
proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
@@ -1806,6 +1832,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
18061832
updatedAt: row.updatedAt,
18071833
archivedAt: row.archivedAt,
18081834
pinnedAt: row.pinnedAt,
1835+
doneOverride: mapThreadDoneOverride(row),
1836+
lastSeenAt: row.lastSeenAt ?? null,
18091837
session: sessionByThread.get(row.threadId) ?? null,
18101838
latestUserMessageAt: row.latestUserMessageAt,
18111839
hasPendingApprovals: row.pendingApprovalCount > 0,
@@ -1955,6 +1983,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
19551983
updatedAt: row.updatedAt,
19561984
archivedAt: row.archivedAt,
19571985
pinnedAt: row.pinnedAt,
1986+
doneOverride: mapThreadDoneOverride(row),
1987+
lastSeenAt: row.lastSeenAt ?? null,
19581988
session: sessionByThread.get(row.threadId) ?? null,
19591989
latestUserMessageAt: row.latestUserMessageAt,
19601990
hasPendingApprovals: row.pendingApprovalCount > 0,
@@ -2221,6 +2251,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
22212251
updatedAt: threadRow.value.updatedAt,
22222252
archivedAt: threadRow.value.archivedAt,
22232253
pinnedAt: threadRow.value.pinnedAt,
2254+
doneOverride: mapThreadDoneOverride(threadRow.value),
2255+
lastSeenAt: threadRow.value.lastSeenAt ?? null,
22242256
session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null,
22252257
latestUserMessageAt: threadRow.value.latestUserMessageAt,
22262258
hasPendingApprovals: threadRow.value.pendingApprovalCount > 0,
@@ -2321,6 +2353,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
23212353
updatedAt: threadRow.value.updatedAt,
23222354
archivedAt: threadRow.value.archivedAt,
23232355
pinnedAt: threadRow.value.pinnedAt,
2356+
doneOverride: mapThreadDoneOverride(threadRow.value),
2357+
lastSeenAt: threadRow.value.lastSeenAt ?? null,
23242358
deletedAt: null,
23252359
messages: messageRows.map(mapThreadMessageRow),
23262360
proposedPlans: proposedPlanRows.map(mapProposedPlanRow),

‎apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,8 @@ function thread(
6464
updatedAt: daysAgo(45),
6565
archivedAt: null,
6666
pinnedAt: null,
67+
doneOverride: null,
68+
lastSeenAt: null,
6769
session: null,
6870
latestUserMessageAt: daysAgo(45),
6971
hasPendingApprovals: false,

‎apps/server/src/orchestration/Schemas.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@ import {
1111
ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema,
1212
ThreadUnarchivedPayload as ContractsThreadUnarchivedPayloadSchema,
1313
ThreadUnpinnedPayload as ContractsThreadUnpinnedPayloadSchema,
14+
ThreadDoneOverrideSetPayload as ContractsThreadDoneOverrideSetPayloadSchema,
15+
ThreadSeenSetPayload as ContractsThreadSeenSetPayloadSchema,
1416
ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema,
1517
ThreadProposedPlanUpsertedPayload as ContractsThreadProposedPlanUpsertedPayloadSchema,
1618
ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema,
@@ -44,6 +46,8 @@ export const ThreadInteractionModeSetPayload = ContractsThreadInteractionModeSet
4446
export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema;
4547
export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema;
4648
export const ThreadUnpinnedPayload = ContractsThreadUnpinnedPayloadSchema;
49+
export const ThreadDoneOverrideSetPayload = ContractsThreadDoneOverrideSetPayloadSchema;
50+
export const ThreadSeenSetPayload = ContractsThreadSeenSetPayloadSchema;
4751

4852
export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema;
4953
export const ThreadProposedPlanUpsertedPayload = ContractsThreadProposedPlanUpsertedPayloadSchema;

‎apps/server/src/orchestration/commandInvariants.test.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,8 @@ const readModel: OrchestrationReadModel = {
7373
updatedAt: now,
7474
archivedAt: null,
7575
pinnedAt: null,
76+
doneOverride: null,
77+
lastSeenAt: null,
7678
latestTurn: null,
7779
messages: [],
7880
session: null,
@@ -100,6 +102,8 @@ const readModel: OrchestrationReadModel = {
100102
updatedAt: now,
101103
archivedAt: null,
102104
pinnedAt: null,
105+
doneOverride: null,
106+
lastSeenAt: null,
103107
latestTurn: null,
104108
messages: [],
105109
session: null,

‎apps/server/src/orchestration/decider.diffStatRebase.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ function makeReadModel(diffStatBaselineTurnCount: number): OrchestrationReadMode
4040
updatedAt: now,
4141
archivedAt: null,
4242
pinnedAt: null,
43+
doneOverride: null,
44+
lastSeenAt: null,
4345
deletedAt: null,
4446
messages: [],
4547
proposedPlans: [],

‎apps/server/src/orchestration/decider.followUp.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,8 @@ const readModelAfterProviderDelivery: OrchestrationReadModel = {
4747
updatedAt: now,
4848
archivedAt: null,
4949
pinnedAt: null,
50+
doneOverride: null,
51+
lastSeenAt: null,
5052
deletedAt: null,
5153
messages: [],
5254
proposedPlans: [],

‎apps/server/src/orchestration/decider.generalChats.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,8 @@ function makeThread(input: {
260260
updatedAt: now,
261261
archivedAt: null,
262262
pinnedAt: null,
263+
doneOverride: null,
264+
lastSeenAt: null,
263265
deletedAt: null,
264266
messages: [
265267
{

0 commit comments

Comments
 (0)