diff --git a/packages/core/src/agent/agent.test.ts b/packages/core/src/agent/agent.test.ts index dda1602..ae4af98 100644 --- a/packages/core/src/agent/agent.test.ts +++ b/packages/core/src/agent/agent.test.ts @@ -276,6 +276,7 @@ describe('buildProjectView', () => { }) it('unbuilt views return a notImplemented marker, not fabricated data', () => { - expect(buildProjectView('standup', undefined, snap, asOf)).toEqual({ notImplemented: 'standup' }) + expect(buildProjectView('milestone', undefined, snap, asOf)).toEqual({ notImplemented: 'milestone' }) + expect(buildProjectView('runway', undefined, snap, asOf)).toEqual({ notImplemented: 'runway' }) }) }) diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 027b7b0..8a2160a 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -25,6 +25,8 @@ export type { export { CACHE_SCHEMA, initCache, readIssue, upsertIssue } from './cache/cache-v0.js' export type { CacheDriver } from './cache/cache-v0.js' +export { affectedIssues, coalesceKey, enqueueWrite, pendingWrites, replayQueue } from './queue/write-queue-v0.js' +export type { QueuedWrite } from './queue/write-queue-v0.js' export { describeChange, isLabelChange, planIssueChange, proposalsFor, summarizeChange } from './changes/apply-changes-v0.js' export type { ChangeProposal, diff --git a/packages/core/src/queue/write-queue-v0.test.ts b/packages/core/src/queue/write-queue-v0.test.ts new file mode 100644 index 0000000..dea486d --- /dev/null +++ b/packages/core/src/queue/write-queue-v0.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it, vi } from 'vitest' + +import type { IssueChange } from '../changes/apply-changes-v0.js' +import { affectedIssues, coalesceKey, enqueueWrite, pendingWrites, replayQueue, type QueuedWrite } from './write-queue-v0.js' + +const reest = (issue: number, estimate: string): IssueChange => + ({ kind: 'reestimate', issue, estimate }) as IssueChange +const assign = (issue: number, assignee: string | null): IssueChange => ({ kind: 'assign', issue, assignee }) + +describe('write-queue-v0 (#33)', () => { + it('coalesces repeat writes to the same (issue, axis) — replay applies the latest once', () => { + let q: QueuedWrite[] = [] + q = enqueueWrite(q, reest(7, 'est/2d'), '2026-02-10T09:00:00Z') + q = enqueueWrite(q, reest(7, 'est/5d'), '2026-02-10T09:05:00Z') // supersedes est/2d + q = enqueueWrite(q, assign(7, 'christian'), '2026-02-10T09:06:00Z') // different axis — kept + + expect(q).toHaveLength(2) + const pending = pendingWrites(q) + expect(pending).toEqual([ + { kind: 'reestimate', issue: 7, estimate: 'est/5d' }, + { kind: 'assign', issue: 7, assignee: 'christian' }, + ]) + expect(coalesceKey(reest(7, 'est/2d'))).toBe('7:reestimate') + }) + + it('keeps writes to different issues and axes distinct', () => { + let q: QueuedWrite[] = [] + q = enqueueWrite(q, reest(7, 'est/2d'), 't1') + q = enqueueWrite(q, reest(8, 'est/3d'), 't2') + q = enqueueWrite(q, assign(8, null), 't3') + expect(q).toHaveLength(3) + expect(affectedIssues(q)).toEqual([7, 8]) + }) + + it('a burst of edits then reconnect lands the final state without duplicating', async () => { + let q: QueuedWrite[] = [] + // offline: three edits to #7's estimate, one assign + q = enqueueWrite(q, reest(7, 'est/1d'), 't1') + q = enqueueWrite(q, reest(7, 'est/2d'), 't2') + q = enqueueWrite(q, reest(7, 'est/8d'), 't3') + q = enqueueWrite(q, assign(7, 'stephen'), 't4') + + const apply = vi.fn(async () => ({ ok: true })) + const { drained, remaining } = await replayQueue(q, apply) + + // only the final estimate + the assign are applied — not three estimate writes + expect(apply).toHaveBeenCalledTimes(2) + expect(apply).toHaveBeenNthCalledWith(1, { kind: 'reestimate', issue: 7, estimate: 'est/8d' }) + expect(apply).toHaveBeenNthCalledWith(2, { kind: 'assign', issue: 7, assignee: 'stephen' }) + expect(remaining).toHaveLength(0) + expect(drained).toHaveLength(2) + }) + + it('keeps writes that still fail on reconnect queued (never throws)', async () => { + let q: QueuedWrite[] = [] + q = enqueueWrite(q, reest(7, 'est/2d'), 't1') + q = enqueueWrite(q, reest(8, 'est/3d'), 't2') + // #7 applies, #8 rejects (still unreachable) — and one apply throws + const apply = vi.fn(async (c: IssueChange) => { + if (c.issue === 8) throw new Error('offline') + return { ok: true } + }) + const { drained, remaining } = await replayQueue(q, apply) + expect(drained.map((w) => w.change.issue)).toEqual([7]) + expect(remaining.map((w) => w.change.issue)).toEqual([8]) + }) +}) diff --git a/packages/core/src/queue/write-queue-v0.ts b/packages/core/src/queue/write-queue-v0.ts new file mode 100644 index 0000000..2a185b6 --- /dev/null +++ b/packages/core/src/queue/write-queue-v0.ts @@ -0,0 +1,71 @@ +/** + * Offline write-queue, v0 (#33). While gitea is unreachable, propose-approved + * writes are queued instead of lost; on reconnect they replay in order and + * resolve against a fresh reconcile. The one hard requirement is *no duplication*: + * replaying must not apply the same intent twice. + * + * The mechanism is coalescing by axis. Every write targets one field of one issue + * (its estimate, priority, assignee, or milestone). Queuing a second write to the + * same (issue, axis) supersedes the first — only the latest intent survives — so + * a burst of edits replays as one final write, and a replay is idempotent (the + * apply path no-ops a change already reflected server-side). + */ + +import type { IssueChange } from '../changes/apply-changes-v0.js' + +export interface QueuedWrite { + change: IssueChange + /** ISO time the write was queued (for display + stable ordering). */ + queuedAt: string +} + +/** + * The coalescing key: one field of one issue. Two writes with the same key are + * the same intent expressed twice — the later wins. Each `IssueChange.kind` maps + * to exactly one axis, so `issue:kind` is the axis identity. + */ +export function coalesceKey(change: IssueChange): string { + return `${change.issue}:${change.kind}` +} + +/** + * Queue a write, superseding any pending write to the same (issue, axis). The + * superseding write moves to the tail so replay order reflects latest intent. + */ +export function enqueueWrite(queue: readonly QueuedWrite[], change: IssueChange, queuedAt: string): QueuedWrite[] { + const key = coalesceKey(change) + return [...queue.filter((w) => coalesceKey(w.change) !== key), { change, queuedAt }] +} + +/** The changes to replay, in order — one per (issue, axis) by construction. */ +export function pendingWrites(queue: readonly QueuedWrite[]): IssueChange[] { + return queue.map((w) => w.change) +} + +/** Distinct issues touched by the queue — what a post-replay reconcile should re-read. */ +export function affectedIssues(queue: readonly QueuedWrite[]): number[] { + return [...new Set(queue.map((w) => w.change.issue))] +} + +/** + * Replay the queue through an apply function (the same guarded write path used + * online), in order. Returns the writes that failed (still unreachable / rejected) + * so they stay queued; everything else drains. Never throws — a failure is data. + */ +export async function replayQueue( + queue: readonly QueuedWrite[], + apply: (change: IssueChange) => Promise<{ ok: boolean }>, +): Promise<{ drained: QueuedWrite[]; remaining: QueuedWrite[] }> { + const drained: QueuedWrite[] = [] + const remaining: QueuedWrite[] = [] + for (const w of queue) { + let ok = false + try { + ok = (await apply(w.change)).ok + } catch { + ok = false + } + ;(ok ? drained : remaining).push(w) + } + return { drained, remaining } +}