Skip to content

Sync inbox lifecycle state across devices #94

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

Merged
badcuban merged 1 commit into from
Jul 31, 2026
Merged

Sync inbox lifecycle state across devices #94

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
81 changes: 81 additions & 0 deletions apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,8 @@ describe("OrchestrationEngine", () => {
updatedAt: "2026-03-03T00:00:03.000Z",
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
deletedAt: null,
messages: [],
proposedPlans: [],
Expand Down Expand Up @@ -485,6 +487,85 @@ describe("OrchestrationEngine", () => {
await system.dispose();
});

it("keeps inbox lifecycle state on the thread so every device reads the same answer", async () => {
const system = await createOrchestrationSystem();
const { engine } = system;
const createdAt = now();

await system.run(
engine.dispatch({
type: "project.create",
commandId: CommandId.make("cmd-project-inbox-create"),
projectId: asProjectId("project-inbox"),
title: "Project Inbox",
workspaceRoot: "/tmp/project-inbox",
defaultModelSelection: {
instanceId: ProviderInstanceId.make("codex"),
model: "gpt-5-codex",
},
createdAt,
}),
);
await system.run(
engine.dispatch({
type: "thread.create",
commandId: CommandId.make("cmd-thread-inbox-create"),
threadId: ThreadId.make("thread-inbox"),
projectId: asProjectId("project-inbox"),
title: "File me",
modelSelection: {
instanceId: ProviderInstanceId.make("codex"),
model: "gpt-5-codex",
},
interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
runtimeMode: "full-access",
branch: null,
worktreePath: null,
createdAt,
}),
);
const readThread = async () =>
(await system.readModel()).threads.find((thread) => thread.id === "thread-inbox");
const updatedAtAfterCreate = (await readThread())?.updatedAt;

await system.run(
engine.dispatch({
type: "thread.done-override.set",
commandId: CommandId.make("cmd-thread-inbox-done"),
threadId: ThreadId.make("thread-inbox"),
state: "done",
at: "2026-01-02T09:30:00.000Z",
}),
);
await system.run(
engine.dispatch({
type: "thread.seen.set",
commandId: CommandId.make("cmd-thread-inbox-seen"),
threadId: ThreadId.make("thread-inbox"),
at: "2026-01-02T09:31:00.000Z",
}),
);
// Mark unread walks the stamp back to just before the completion it
// un-sees, so a later write with an earlier stamp has to win.
await system.run(
engine.dispatch({
type: "thread.seen.set",
commandId: CommandId.make("cmd-thread-inbox-unread"),
threadId: ThreadId.make("thread-inbox"),
at: "2026-01-02T09:29:59.999Z",
}),
);

const thread = await readThread();
expect(thread?.doneOverride).toEqual({ state: "done", at: "2026-01-02T09:30:00.000Z" });
expect(thread?.lastSeenAt).toBe("2026-01-02T09:29:59.999Z");
// Filing and reading are not work: `updatedAt` is what the inbox weighs
// these stamps against, and what auto-wrap idles off.
expect(thread?.updatedAt).toBe(updatedAtAfterCreate);

await system.dispose();
});

it("replays append-only events from sequence", async () => {
const system = await createOrchestrationSystem();
const { engine } = system;
Expand Down
34 changes: 34 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -565,6 +565,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
updatedAt: event.payload.updatedAt,
archivedAt: null,
pinnedAt: null,
doneOverride: null,
doneOverrideAt: null,
lastSeenAt: null,
latestUserMessageAt: null,
pendingApprovalCount: 0,
pendingUserInputCount: 0,
Expand Down Expand Up @@ -633,6 +636,37 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
return;
}

// Inbox filing and read state deliberately leave `updatedAt` alone --
// the client weighs both stamps against the thread's real activity.
case "thread.done-override-set": {
const existingRow = yield* projectionThreadRepository.getById({
threadId: event.payload.threadId,
});
if (Option.isNone(existingRow)) {
return;
}
yield* projectionThreadRepository.upsert({
...existingRow.value,
doneOverride: event.payload.state,
doneOverrideAt: event.payload.at,
});
return;
}

case "thread.seen-set": {
const existingRow = yield* projectionThreadRepository.getById({
threadId: event.payload.threadId,
});
if (Option.isNone(existingRow)) {
return;
}
yield* projectionThreadRepository.upsert({
...existingRow.value,
lastSeenAt: event.payload.at,
});
return;
}

case "thread.meta-updated": {
const existingRow = yield* projectionThreadRepository.getById({
threadId: event.payload.threadId,
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
updatedAt: "2026-02-24T00:00:03.000Z",
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
deletedAt: null,
messages: [
{
Expand Down Expand Up @@ -435,6 +437,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
updatedAt: "2026-02-24T00:00:03.000Z",
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
session: {
threadId: ThreadId.make("thread-1"),
status: "running",
Expand Down
34 changes: 34 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import {
type OrchestrationSession,
type OrchestrationThreadActivity,
type OrchestrationThreadDiffStat,
type OrchestrationThreadDoneOverride,
type OrchestrationThreadShell,
ModelSelection,
OrchestrationThreadGoal,
Expand Down Expand Up @@ -231,6 +232,15 @@ function mapLatestTurn(
};
}

function mapThreadDoneOverride(
row: Schema.Schema.Type<typeof ProjectionThreadDbRowSchema>,
): OrchestrationThreadDoneOverride | null {
if (row.doneOverride == null || row.doneOverrideAt == null) {
return null;
}
return { state: row.doneOverride, at: row.doneOverrideAt };
}

function mapThreadDiffStat(
row: Schema.Schema.Type<typeof ProjectionThreadDiffStatDbRowSchema> | undefined,
): OrchestrationThreadDiffStat | null {
Expand Down Expand Up @@ -392,6 +402,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updated_at AS "updatedAt",
archived_at AS "archivedAt",
pinned_at AS "pinnedAt",
done_override AS "doneOverride",
done_override_at AS "doneOverrideAt",
last_seen_at AS "lastSeenAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
Expand Down Expand Up @@ -425,6 +438,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updated_at AS "updatedAt",
archived_at AS "archivedAt",
pinned_at AS "pinnedAt",
done_override AS "doneOverride",
done_override_at AS "doneOverrideAt",
last_seen_at AS "lastSeenAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
Expand Down Expand Up @@ -460,6 +476,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updated_at AS "updatedAt",
archived_at AS "archivedAt",
pinned_at AS "pinnedAt",
done_override AS "doneOverride",
done_override_at AS "doneOverrideAt",
last_seen_at AS "lastSeenAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
Expand Down Expand Up @@ -977,6 +996,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updated_at AS "updatedAt",
archived_at AS "archivedAt",
pinned_at AS "pinnedAt",
done_override AS "doneOverride",
done_override_at AS "doneOverrideAt",
last_seen_at AS "lastSeenAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
Expand Down Expand Up @@ -1423,6 +1445,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
pinnedAt: row.pinnedAt,
doneOverride: mapThreadDoneOverride(row),
lastSeenAt: row.lastSeenAt ?? null,
deletedAt: row.deletedAt,
messages: messagesByThread.get(row.threadId) ?? [],
proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
Expand Down Expand Up @@ -1660,6 +1684,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
pinnedAt: row.pinnedAt,
doneOverride: mapThreadDoneOverride(row),
lastSeenAt: row.lastSeenAt ?? null,
deletedAt: row.deletedAt,
messages: messagesByThread.get(row.threadId) ?? [],
proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
Expand Down Expand Up @@ -1806,6 +1832,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
pinnedAt: row.pinnedAt,
doneOverride: mapThreadDoneOverride(row),
lastSeenAt: row.lastSeenAt ?? null,
session: sessionByThread.get(row.threadId) ?? null,
latestUserMessageAt: row.latestUserMessageAt,
hasPendingApprovals: row.pendingApprovalCount > 0,
Expand Down Expand Up @@ -1955,6 +1983,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
pinnedAt: row.pinnedAt,
doneOverride: mapThreadDoneOverride(row),
lastSeenAt: row.lastSeenAt ?? null,
session: sessionByThread.get(row.threadId) ?? null,
latestUserMessageAt: row.latestUserMessageAt,
hasPendingApprovals: row.pendingApprovalCount > 0,
Expand Down Expand Up @@ -2221,6 +2251,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updatedAt: threadRow.value.updatedAt,
archivedAt: threadRow.value.archivedAt,
pinnedAt: threadRow.value.pinnedAt,
doneOverride: mapThreadDoneOverride(threadRow.value),
lastSeenAt: threadRow.value.lastSeenAt ?? null,
session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null,
latestUserMessageAt: threadRow.value.latestUserMessageAt,
hasPendingApprovals: threadRow.value.pendingApprovalCount > 0,
Expand Down Expand Up @@ -2321,6 +2353,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
updatedAt: threadRow.value.updatedAt,
archivedAt: threadRow.value.archivedAt,
pinnedAt: threadRow.value.pinnedAt,
doneOverride: mapThreadDoneOverride(threadRow.value),
lastSeenAt: threadRow.value.lastSeenAt ?? null,
deletedAt: null,
messages: messageRows.map(mapThreadMessageRow),
proposedPlans: proposedPlanRows.map(mapProposedPlanRow),
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,8 @@ function thread(
updatedAt: daysAgo(45),
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
session: null,
latestUserMessageAt: daysAgo(45),
hasPendingApprovals: false,
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/orchestration/Schemas.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@ import {
ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema,
ThreadUnarchivedPayload as ContractsThreadUnarchivedPayloadSchema,
ThreadUnpinnedPayload as ContractsThreadUnpinnedPayloadSchema,
ThreadDoneOverrideSetPayload as ContractsThreadDoneOverrideSetPayloadSchema,
ThreadSeenSetPayload as ContractsThreadSeenSetPayloadSchema,
ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema,
ThreadProposedPlanUpsertedPayload as ContractsThreadProposedPlanUpsertedPayloadSchema,
ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema,
Expand Down Expand Up @@ -44,6 +46,8 @@ export const ThreadInteractionModeSetPayload = ContractsThreadInteractionModeSet
export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema;
export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema;
export const ThreadUnpinnedPayload = ContractsThreadUnpinnedPayloadSchema;
export const ThreadDoneOverrideSetPayload = ContractsThreadDoneOverrideSetPayloadSchema;
export const ThreadSeenSetPayload = ContractsThreadSeenSetPayloadSchema;

export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema;
export const ThreadProposedPlanUpsertedPayload = ContractsThreadProposedPlanUpsertedPayloadSchema;
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/orchestration/commandInvariants.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,8 @@ const readModel: OrchestrationReadModel = {
updatedAt: now,
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
latestTurn: null,
messages: [],
session: null,
Expand Down Expand Up @@ -100,6 +102,8 @@ const readModel: OrchestrationReadModel = {
updatedAt: now,
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
latestTurn: null,
messages: [],
session: null,
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration/decider.diffStatRebase.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ function makeReadModel(diffStatBaselineTurnCount: number): OrchestrationReadMode
updatedAt: now,
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
deletedAt: null,
messages: [],
proposedPlans: [],
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration/decider.followUp.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,8 @@ const readModelAfterProviderDelivery: OrchestrationReadModel = {
updatedAt: now,
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
deletedAt: null,
messages: [],
proposedPlans: [],
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration/decider.generalChats.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,8 @@ function makeThread(input: {
updatedAt: now,
archivedAt: null,
pinnedAt: null,
doneOverride: null,
lastSeenAt: null,
deletedAt: null,
messages: [
{
Expand Down
Loading
Loading