feat(member-work-sync): dispatch nudges after reconcile
This commit is contained in:
parent
23714f5ca8
commit
895c3afc5e
4 changed files with 149 additions and 1 deletions
|
|
@ -36,7 +36,7 @@ Current implementation note:
|
|||
- 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.
|
||||
- Dispatcher use case exists behind explicit facade invocation. It claims due outbox rows, revalidates active team/status/current fingerprint/readiness/busy/watchdog cooldown, then writes one idempotent inbox nudge through a narrow port.
|
||||
- Dispatcher use case runs after queued reconcile and is also exposed through the facade. It claims due outbox rows, revalidates active team/status/current fingerprint/readiness/busy/watchdog cooldown, then writes one idempotent inbox nudge through a narrow port.
|
||||
- Phase 2 must not start until real shadow metrics confirm that `needs_sync` churn and false positives are acceptably low.
|
||||
|
||||
Patterns used:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,63 @@
|
|||
import { readFile } from 'fs/promises';
|
||||
import { join } from 'path';
|
||||
import type { MemberWorkSyncWatchdogCooldownPort } from '../../../core/application';
|
||||
|
||||
const DEFAULT_WATCHDOG_COOLDOWN_MS = 10 * 60_000;
|
||||
|
||||
interface StallJournalEntry {
|
||||
taskId: string;
|
||||
state: string;
|
||||
alertedAt?: string;
|
||||
}
|
||||
|
||||
function parseTime(value: string | undefined): number | null {
|
||||
if (!value) {
|
||||
return null;
|
||||
}
|
||||
const time = new Date(value).getTime();
|
||||
return Number.isFinite(time) ? time : null;
|
||||
}
|
||||
|
||||
export class TeamTaskStallJournalWorkSyncCooldown implements MemberWorkSyncWatchdogCooldownPort {
|
||||
constructor(
|
||||
private readonly teamsBasePath: string,
|
||||
private readonly cooldownMs: number = DEFAULT_WATCHDOG_COOLDOWN_MS
|
||||
) {}
|
||||
|
||||
async hasRecentNudge(input: {
|
||||
teamName: string;
|
||||
memberName: string;
|
||||
taskIds: string[];
|
||||
nowIso: string;
|
||||
}): Promise<boolean> {
|
||||
const taskIds = new Set(input.taskIds);
|
||||
if (taskIds.size === 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
const raw = await readFile(
|
||||
join(this.teamsBasePath, input.teamName, 'stall-monitor-journal.json'),
|
||||
'utf8'
|
||||
);
|
||||
const parsed = JSON.parse(raw) as unknown;
|
||||
if (!Array.isArray(parsed)) {
|
||||
return true;
|
||||
}
|
||||
const now = parseTime(input.nowIso) ?? Date.now();
|
||||
return parsed.some((entry): boolean => {
|
||||
const row = entry as Partial<StallJournalEntry>;
|
||||
if (row.state !== 'alerted' || !row.taskId || !taskIds.has(row.taskId)) {
|
||||
return false;
|
||||
}
|
||||
const alertedAt = parseTime(row.alertedAt);
|
||||
return alertedAt != null && now - alertedAt <= this.cooldownMs;
|
||||
});
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -19,6 +19,7 @@ import {
|
|||
} from '../../core/application';
|
||||
import { MemberWorkSyncTeamChangeRouter } from '../adapters/input/MemberWorkSyncTeamChangeRouter';
|
||||
import { TeamInboxMemberWorkSyncNudgeSink } from '../adapters/output/TeamInboxMemberWorkSyncNudgeSink';
|
||||
import { TeamTaskStallJournalWorkSyncCooldown } from '../adapters/output/TeamTaskStallJournalWorkSyncCooldown';
|
||||
import { TeamTaskAgendaSource } from '../adapters/output/TeamTaskAgendaSource';
|
||||
import { HmacMemberWorkSyncReportTokenAdapter } from '../infrastructure/HmacMemberWorkSyncReportTokenAdapter';
|
||||
import {
|
||||
|
|
@ -72,6 +73,7 @@ export function createMemberWorkSyncFeature(deps: {
|
|||
const store = new JsonMemberWorkSyncStore(storePaths);
|
||||
const reportToken = new HmacMemberWorkSyncReportTokenAdapter(storePaths);
|
||||
const inboxNudge = new TeamInboxMemberWorkSyncNudgeSink();
|
||||
const watchdogCooldown = new TeamTaskStallJournalWorkSyncCooldown(deps.teamsBasePath);
|
||||
const useCaseDeps = {
|
||||
clock,
|
||||
hash,
|
||||
|
|
@ -80,6 +82,7 @@ export function createMemberWorkSyncFeature(deps: {
|
|||
reportStore: store,
|
||||
outboxStore: store,
|
||||
inboxNudge,
|
||||
watchdogCooldown,
|
||||
reportToken,
|
||||
...(deps.isTeamActive ? { lifecycle: { isTeamActive: deps.isTeamActive } } : {}),
|
||||
logger: deps.logger,
|
||||
|
|
@ -93,6 +96,10 @@ export function createMemberWorkSyncFeature(deps: {
|
|||
const queue = new MemberWorkSyncEventQueue({
|
||||
reconcile: async (request, context: MemberWorkSyncReconcileContext) => {
|
||||
await reconciler.execute(request, context);
|
||||
await nudgeDispatcher.dispatchDue({
|
||||
teamNames: [request.teamName],
|
||||
claimedBy: `member-work-sync:${process.pid}`,
|
||||
});
|
||||
},
|
||||
isTeamActive: deps.isTeamActive ?? (() => true),
|
||||
logger: deps.logger,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,78 @@
|
|||
import { mkdir, mkdtemp, rm, writeFile } from 'fs/promises';
|
||||
import { tmpdir } from 'os';
|
||||
import { join } from 'path';
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
|
||||
|
||||
import { TeamTaskStallJournalWorkSyncCooldown } from '@features/member-work-sync/main/adapters/output/TeamTaskStallJournalWorkSyncCooldown';
|
||||
|
||||
describe('TeamTaskStallJournalWorkSyncCooldown', () => {
|
||||
let root: string;
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'member-work-sync-watchdog-cooldown-'));
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('detects recent watchdog alerts for the same task', async () => {
|
||||
await mkdir(join(root, 'team-a'), { recursive: true });
|
||||
await writeFile(
|
||||
join(root, 'team-a', 'stall-monitor-journal.json'),
|
||||
JSON.stringify([
|
||||
{
|
||||
taskId: 'task-1',
|
||||
state: 'alerted',
|
||||
alertedAt: '2026-04-29T00:05:00.000Z',
|
||||
},
|
||||
]),
|
||||
'utf8'
|
||||
);
|
||||
|
||||
const cooldown = new TeamTaskStallJournalWorkSyncCooldown(root, 10 * 60_000);
|
||||
|
||||
await expect(
|
||||
cooldown.hasRecentNudge({
|
||||
teamName: 'team-a',
|
||||
memberName: 'bob',
|
||||
taskIds: ['task-1'],
|
||||
nowIso: '2026-04-29T00:10:00.000Z',
|
||||
})
|
||||
).resolves.toBe(true);
|
||||
});
|
||||
|
||||
it('ignores old watchdog alerts and missing journals', async () => {
|
||||
await mkdir(join(root, 'team-a'), { recursive: true });
|
||||
await writeFile(
|
||||
join(root, 'team-a', 'stall-monitor-journal.json'),
|
||||
JSON.stringify([
|
||||
{
|
||||
taskId: 'task-1',
|
||||
state: 'alerted',
|
||||
alertedAt: '2026-04-29T00:00:00.000Z',
|
||||
},
|
||||
]),
|
||||
'utf8'
|
||||
);
|
||||
|
||||
const cooldown = new TeamTaskStallJournalWorkSyncCooldown(root, 10 * 60_000);
|
||||
|
||||
await expect(
|
||||
cooldown.hasRecentNudge({
|
||||
teamName: 'team-a',
|
||||
memberName: 'bob',
|
||||
taskIds: ['task-1'],
|
||||
nowIso: '2026-04-29T00:20:00.000Z',
|
||||
})
|
||||
).resolves.toBe(false);
|
||||
await expect(
|
||||
cooldown.hasRecentNudge({
|
||||
teamName: 'team-missing',
|
||||
memberName: 'bob',
|
||||
taskIds: ['task-1'],
|
||||
nowIso: '2026-04-29T00:20:00.000Z',
|
||||
})
|
||||
).resolves.toBe(false);
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue