From 3552a4ba61fd08b1b2cf5f8d634cf3d743ccf183 Mon Sep 17 00:00:00 2001 From: 777genius Date: Wed, 29 Apr 2026 15:20:51 +0300 Subject: [PATCH] feat(member-work-sync): plan nudges behind readiness --- .../member-work-sync-control-plane-plan.md | 1 + .../MemberWorkSyncNudgeOutboxPlanner.ts | 56 +++++++++ .../application/MemberWorkSyncReconciler.ts | 20 ++- .../core/application/index.ts | 1 + .../core/domain/MemberWorkSyncNudge.ts | 105 ++++++++++++++++ .../member-work-sync/core/domain/index.ts | 1 + .../createMemberWorkSyncFeature.ts | 1 + .../core/MemberWorkSyncUseCases.test.ts | 118 ++++++++++++++++++ 8 files changed, 302 insertions(+), 1 deletion(-) create mode 100644 src/features/member-work-sync/core/application/MemberWorkSyncNudgeOutboxPlanner.ts create mode 100644 src/features/member-work-sync/core/domain/MemberWorkSyncNudge.ts diff --git a/docs/team-management/member-work-sync-control-plane-plan.md b/docs/team-management/member-work-sync-control-plane-plan.md index 14f64d56..9d51cfd9 100644 --- a/docs/team-management/member-work-sync-control-plane-plan.md +++ b/docs/team-management/member-work-sync-control-plane-plan.md @@ -35,6 +35,7 @@ Current implementation note: - Phase 1 does not insert inbox messages, send nudges, mark tasks/messages read, or change `TeamTaskStallMonitor` semantics. - Phase 1.5 exposes a machine-readable `phase2Readiness` assessment from shadow metrics. It can say `collecting_shadow_data`, `blocked`, or `shadow_ready`; it still does not dispatch nudges. - Phase 2 storage foundation is implemented as an inert durable outbox: idempotency key, payload hash conflict checks, claim generation fencing, retry/terminal states. It is not wired to dispatch inbox nudges yet. +- Reconciler can plan a Phase 2 outbox item only when `phase2Readiness=shadow_ready`; otherwise it records normal shadow status and does nothing. This preserves the anti-spam gate before any dispatcher exists. - Phase 2 must not start until real shadow metrics confirm that `needs_sync` churn and false positives are acceptably low. Patterns used: diff --git a/src/features/member-work-sync/core/application/MemberWorkSyncNudgeOutboxPlanner.ts b/src/features/member-work-sync/core/application/MemberWorkSyncNudgeOutboxPlanner.ts new file mode 100644 index 00000000..8281a87a --- /dev/null +++ b/src/features/member-work-sync/core/application/MemberWorkSyncNudgeOutboxPlanner.ts @@ -0,0 +1,56 @@ +import type { MemberWorkSyncStatus } from '../../contracts'; +import { buildMemberWorkSyncOutboxEnsureInput } from '../domain'; +import type { MemberWorkSyncUseCaseDeps } from './ports'; + +export interface MemberWorkSyncNudgeOutboxPlanResult { + planned: boolean; + code: + | 'outbox_unavailable' + | 'metrics_unavailable' + | 'status_not_nudgeable' + | 'phase2_not_ready' + | 'created' + | 'existing' + | 'payload_conflict'; +} + +export class MemberWorkSyncNudgeOutboxPlanner { + constructor(private readonly deps: MemberWorkSyncUseCaseDeps) {} + + async plan(status: MemberWorkSyncStatus): Promise { + if (!this.deps.outboxStore) { + return { planned: false, code: 'outbox_unavailable' }; + } + if (!this.deps.statusStore.readTeamMetrics) { + return { planned: false, code: 'metrics_unavailable' }; + } + + const input = buildMemberWorkSyncOutboxEnsureInput({ + status, + hash: this.deps.hash, + nowIso: status.evaluatedAt, + }); + if (!input) { + return { planned: false, code: 'status_not_nudgeable' }; + } + + const metrics = await this.deps.statusStore.readTeamMetrics(status.teamName); + if (metrics.phase2Readiness.state !== 'shadow_ready') { + return { planned: false, code: 'phase2_not_ready' }; + } + + const result = await this.deps.outboxStore.ensurePending(input); + if (!result.ok) { + this.deps.logger?.warn('member work sync nudge outbox payload conflict', { + teamName: status.teamName, + memberName: status.memberName, + outboxId: input.id, + existingPayloadHash: result.existingPayloadHash, + requestedPayloadHash: result.requestedPayloadHash, + }); + return { planned: false, code: 'payload_conflict' }; + } + + return { planned: true, code: result.outcome }; + } +} diff --git a/src/features/member-work-sync/core/application/MemberWorkSyncReconciler.ts b/src/features/member-work-sync/core/application/MemberWorkSyncReconciler.ts index a98d9cf6..1459d06e 100644 --- a/src/features/member-work-sync/core/application/MemberWorkSyncReconciler.ts +++ b/src/features/member-work-sync/core/application/MemberWorkSyncReconciler.ts @@ -6,6 +6,7 @@ import { } from '../domain'; import type { MemberWorkSyncStatus, MemberWorkSyncStatusRequest } from '../../contracts'; import type { MemberWorkSyncAgendaSourceResult, MemberWorkSyncUseCaseDeps } from './ports'; +import { MemberWorkSyncNudgeOutboxPlanner } from './MemberWorkSyncNudgeOutboxPlanner'; export interface MemberWorkSyncReconcileContext { reconciledBy?: 'request' | 'queue'; @@ -33,7 +34,11 @@ export function finalizeMemberWorkSyncAgenda( } export class MemberWorkSyncReconciler { - constructor(private readonly deps: MemberWorkSyncUseCaseDeps) {} + private readonly nudgeOutboxPlanner: MemberWorkSyncNudgeOutboxPlanner; + + constructor(private readonly deps: MemberWorkSyncUseCaseDeps) { + this.nudgeOutboxPlanner = new MemberWorkSyncNudgeOutboxPlanner(deps); + } async execute( request: MemberWorkSyncStatusRequest, @@ -82,8 +87,21 @@ export class MemberWorkSyncReconciler { }); await this.deps.statusStore.write(status); + await this.planNudgeOutbox(status); return status; } + + private async planNudgeOutbox(status: MemberWorkSyncStatus): Promise { + const result = await this.nudgeOutboxPlanner.plan(status); + if (result.code !== 'outbox_unavailable' && result.code !== 'status_not_nudgeable') { + this.deps.logger?.debug('member work sync nudge outbox planning result', { + teamName: status.teamName, + memberName: status.memberName, + code: result.code, + planned: result.planned, + }); + } + } } export async function attachMemberWorkSyncReportToken( diff --git a/src/features/member-work-sync/core/application/index.ts b/src/features/member-work-sync/core/application/index.ts index b47b511c..9391fad3 100644 --- a/src/features/member-work-sync/core/application/index.ts +++ b/src/features/member-work-sync/core/application/index.ts @@ -1,5 +1,6 @@ export * from './MemberWorkSyncDiagnosticsReader'; export * from './MemberWorkSyncMetricsReader'; +export * from './MemberWorkSyncNudgeOutboxPlanner'; export * from './MemberWorkSyncPendingReportIntentReplayer'; export * from './MemberWorkSyncReconciler'; export * from './MemberWorkSyncReporter'; diff --git a/src/features/member-work-sync/core/domain/MemberWorkSyncNudge.ts b/src/features/member-work-sync/core/domain/MemberWorkSyncNudge.ts new file mode 100644 index 00000000..4b48a074 --- /dev/null +++ b/src/features/member-work-sync/core/domain/MemberWorkSyncNudge.ts @@ -0,0 +1,105 @@ +import type { + MemberWorkSyncNudgePayload, + MemberWorkSyncOutboxEnsureInput, + MemberWorkSyncStatus, +} from '../../contracts'; + +export const MEMBER_WORK_SYNC_NUDGE_ID_PREFIX = 'member-work-sync'; + +interface MemberWorkSyncNudgeHash { + sha256Hex(value: string): string; +} + +function stableJson(value: unknown): string { + if (value == null || typeof value !== 'object') { + return JSON.stringify(value); + } + if (Array.isArray(value)) { + return `[${value.map(stableJson).join(',')}]`; + } + + const record = value as Record; + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${stableJson(record[key])}`) + .join(',')}}`; +} + +export function buildMemberWorkSyncNudgeId(input: { + teamName: string; + memberName: string; + agendaFingerprint: string; +}): string { + return [ + MEMBER_WORK_SYNC_NUDGE_ID_PREFIX, + input.teamName, + input.memberName.trim().toLowerCase(), + input.agendaFingerprint, + ].join(':'); +} + +export function buildMemberWorkSyncNudgePayload(status: MemberWorkSyncStatus): MemberWorkSyncNudgePayload { + const taskRefs = status.agenda.items.map((item) => ({ + teamName: status.teamName, + taskId: item.taskId, + displayId: item.displayId ?? item.taskId.slice(0, 8), + })); + const preview = status.agenda.items + .slice(0, 3) + .map((item) => `${item.displayId ?? item.taskId.slice(0, 8)} ${item.subject}`) + .join('; '); + + return { + from: 'system', + to: status.memberName, + messageKind: 'member_work_sync_nudge', + source: 'member-work-sync', + actionMode: 'do', + taskRefs, + text: [ + 'Work sync check: you have current actionable work assigned.', + preview ? `Current agenda: ${preview}.` : '', + 'Continue concrete task work, report a real blocker with task tools, or call member_work_sync_report for the current fingerprint.', + 'Do not reply only with acknowledgement.', + ] + .filter(Boolean) + .join('\n'), + }; +} + +export function buildMemberWorkSyncNudgePayloadHash( + hash: MemberWorkSyncNudgeHash, + payload: MemberWorkSyncNudgePayload +): string { + return hash.sha256Hex(stableJson(payload)); +} + +export function buildMemberWorkSyncOutboxEnsureInput(input: { + status: MemberWorkSyncStatus; + hash: MemberWorkSyncNudgeHash; + nowIso: string; +}): MemberWorkSyncOutboxEnsureInput | null { + const status = input.status; + if ( + status.state !== 'needs_sync' || + status.shadow?.wouldNudge !== true || + status.agenda.items.length === 0 + ) { + return null; + } + + const payload = buildMemberWorkSyncNudgePayload(status); + return { + id: buildMemberWorkSyncNudgeId({ + teamName: status.teamName, + memberName: status.memberName, + agendaFingerprint: status.agenda.fingerprint, + }), + teamName: status.teamName, + memberName: status.memberName, + agendaFingerprint: status.agenda.fingerprint, + payloadHash: buildMemberWorkSyncNudgePayloadHash(input.hash, payload), + payload, + nowIso: input.nowIso, + }; +} diff --git a/src/features/member-work-sync/core/domain/index.ts b/src/features/member-work-sync/core/domain/index.ts index 11ba63d6..2925168c 100644 --- a/src/features/member-work-sync/core/domain/index.ts +++ b/src/features/member-work-sync/core/domain/index.ts @@ -3,5 +3,6 @@ export * from './AgendaFingerprint'; export * from './currentReviewCycle'; export * from './memberName'; export * from './MemberWorkSyncPhase2Readiness'; +export * from './MemberWorkSyncNudge'; export * from './MemberWorkSyncReportValidator'; export * from './SyncDecisionPolicy'; diff --git a/src/features/member-work-sync/main/composition/createMemberWorkSyncFeature.ts b/src/features/member-work-sync/main/composition/createMemberWorkSyncFeature.ts index 40034a4e..ce76d27f 100644 --- a/src/features/member-work-sync/main/composition/createMemberWorkSyncFeature.ts +++ b/src/features/member-work-sync/main/composition/createMemberWorkSyncFeature.ts @@ -73,6 +73,7 @@ export function createMemberWorkSyncFeature(deps: { agendaSource, statusStore: store, reportStore: store, + outboxStore: store, reportToken, ...(deps.isTeamActive ? { lifecycle: { isTeamActive: deps.isTeamActive } } : {}), logger: deps.logger, diff --git a/test/features/member-work-sync/core/MemberWorkSyncUseCases.test.ts b/test/features/member-work-sync/core/MemberWorkSyncUseCases.test.ts index 18fea140..23060c3d 100644 --- a/test/features/member-work-sync/core/MemberWorkSyncUseCases.test.ts +++ b/test/features/member-work-sync/core/MemberWorkSyncUseCases.test.ts @@ -5,14 +5,22 @@ import { MemberWorkSyncPendingReportIntentReplayer, MemberWorkSyncReporter, type MemberWorkSyncAgendaSourceResult, + type MemberWorkSyncOutboxStorePort, type MemberWorkSyncStatusStorePort, type MemberWorkSyncUseCaseDeps, } from '@features/member-work-sync/core/application'; import type { MemberWorkSyncActionableWorkItem, + MemberWorkSyncOutboxEnsureInput, + MemberWorkSyncOutboxItem, + MemberWorkSyncOutboxMarkDeliveredInput, + MemberWorkSyncOutboxMarkFailedInput, + MemberWorkSyncOutboxMarkSupersededInput, + MemberWorkSyncPhase2ReadinessState, MemberWorkSyncReportIntent, MemberWorkSyncReportRequest, MemberWorkSyncStatus, + MemberWorkSyncTeamMetrics, } from '@features/member-work-sync/contracts'; const workItem: MemberWorkSyncActionableWorkItem = { @@ -45,6 +53,7 @@ class InMemoryStatusStore implements MemberWorkSyncStatusStorePort { readonly writes: MemberWorkSyncStatus[] = []; readonly pendingReports: Array<{ request: MemberWorkSyncReportRequest; reason: string }> = []; readonly pendingIntents = new Map(); + phase2ReadinessState: MemberWorkSyncPhase2ReadinessState = 'collecting_shadow_data'; async read(): Promise { return this.writes.at(-1) ?? null; @@ -76,6 +85,74 @@ class InMemoryStatusStore implements MemberWorkSyncStatusStorePort { this.pendingIntents.set(id, { ...current, ...result }); } } + + async readTeamMetrics(teamName: string): Promise { + return { + teamName, + generatedAt: '2026-04-29T00:00:00.000Z', + memberCount: 1, + stateCounts: { + caught_up: 0, + needs_sync: 1, + still_working: 0, + blocked: 0, + inactive: 0, + unknown: 0, + }, + actionableItemCount: this.writes.at(-1)?.agenda.items.length ?? 0, + wouldNudgeCount: 1, + fingerprintChangeCount: 0, + reportAcceptedCount: 0, + reportRejectedCount: 0, + recentEvents: [], + phase2Readiness: { + state: this.phase2ReadinessState, + reasons: [], + thresholds: { + minObservedMembers: 1, + minStatusEvents: 20, + minObservationHours: 1, + maxWouldNudgesPerMemberHour: 2, + maxFingerprintChangesPerMemberHour: 1, + maxReportRejectionRate: 0.2, + }, + rates: { + observationHours: 2, + statusEventCount: 30, + wouldNudgesPerMemberHour: 0.5, + fingerprintChangesPerMemberHour: 0, + reportRejectionRate: 0, + }, + diagnostics: [], + }, + }; + } +} + +class InMemoryOutboxStore implements MemberWorkSyncOutboxStorePort { + readonly ensures: MemberWorkSyncOutboxEnsureInput[] = []; + + async ensurePending(input: MemberWorkSyncOutboxEnsureInput) { + this.ensures.push(input); + const item: MemberWorkSyncOutboxItem = { + ...input, + status: 'pending', + attemptGeneration: 0, + createdAt: input.nowIso, + updatedAt: input.nowIso, + }; + return { ok: true as const, outcome: 'created' as const, item }; + } + + async claimDue(): Promise { + return []; + } + + async markDelivered(_input: MemberWorkSyncOutboxMarkDeliveredInput): Promise {} + + async markSuperseded(_input: MemberWorkSyncOutboxMarkSupersededInput): Promise {} + + async markFailed(_input: MemberWorkSyncOutboxMarkFailedInput): Promise {} } function createDeps(options?: { @@ -84,6 +161,7 @@ function createDeps(options?: { inactive?: boolean; teamActive?: boolean; providerId?: 'opencode' | 'codex'; + outboxStore?: MemberWorkSyncOutboxStorePort; }) { const clock = new MutableClock(); const store = new InMemoryStatusStore(); @@ -110,6 +188,7 @@ function createDeps(options?: { }, statusStore: store, reportStore: store, + ...(options?.outboxStore ? { outboxStore: options.outboxStore } : {}), reportToken: { create: async (input) => ({ token: `token:${input.teamName}:${input.memberName}:${input.agendaFingerprint}`, @@ -277,6 +356,45 @@ describe('MemberWorkSync use cases', () => { expect(changed.state).toBe('needs_sync'); }); + it('does not create outbox nudges until shadow readiness is green', async () => { + const outbox = new InMemoryOutboxStore(); + const { deps } = createDeps({ outboxStore: outbox }); + + await new MemberWorkSyncDiagnosticsReader(deps).execute({ + teamName: 'team-a', + memberName: 'bob', + }); + + expect(outbox.ensures).toEqual([]); + }); + + it('creates one idempotent outbox nudge intent when Phase 2 readiness is green', async () => { + const outbox = new InMemoryOutboxStore(); + const { deps, store } = createDeps({ outboxStore: outbox }); + store.phase2ReadinessState = 'shadow_ready'; + + const status = await new MemberWorkSyncDiagnosticsReader(deps).execute({ + teamName: 'team-a', + memberName: 'bob', + }); + + expect(outbox.ensures).toHaveLength(1); + expect(outbox.ensures[0]).toMatchObject({ + id: `member-work-sync:team-a:bob:${status.agenda.fingerprint}`, + teamName: 'team-a', + memberName: 'bob', + agendaFingerprint: status.agenda.fingerprint, + payload: { + from: 'system', + to: 'bob', + messageKind: 'member_work_sync_nudge', + source: 'member-work-sync', + actionMode: 'do', + taskRefs: [{ teamName: 'team-a', taskId: 'task-1', displayId: '11111111' }], + }, + }); + }); + it('rejects invalid report tokens without recording replayable intents', async () => { const { deps, store } = createDeps(); const reader = new MemberWorkSyncDiagnosticsReader(deps);