diff --git a/src/features/member-work-sync/core/application/RuntimeTurnSettledIngestor.ts b/src/features/member-work-sync/core/application/RuntimeTurnSettledIngestor.ts index 4dc6ee5a..a745557d 100644 --- a/src/features/member-work-sync/core/application/RuntimeTurnSettledIngestor.ts +++ b/src/features/member-work-sync/core/application/RuntimeTurnSettledIngestor.ts @@ -1,4 +1,5 @@ import type { MemberWorkSyncClockPort, MemberWorkSyncLoggerPort } from './ports'; +import type { RuntimeTurnSettledEvent } from '../domain'; import type { RuntimeTurnSettledEventStorePort, RuntimeTurnSettledPayloadNormalizerPort, @@ -19,10 +20,30 @@ export interface RuntimeTurnSettledDrainSummary { claimed: number; enqueued: number; unresolved: number; + ignored: number; invalid: number; failed: number; } +const NON_TERMINAL_OPENCODE_OUTCOMES = new Set([ + 'timeout', + 'stream_unavailable', + 'prompt_rejected', + 'idle_without_assistant_activity', + 'unknown', +]); + +function getIgnoredReason(event: RuntimeTurnSettledEvent): string | null { + if (event.provider !== 'opencode') { + return null; + } + const outcome = event.outcome?.trim(); + if (!outcome || !NON_TERMINAL_OPENCODE_OUTCOMES.has(outcome)) { + return null; + } + return `opencode_non_terminal_outcome:${outcome}`; +} + export class RuntimeTurnSettledIngestor { constructor(private readonly deps: RuntimeTurnSettledIngestorDeps) {} @@ -31,6 +52,7 @@ export class RuntimeTurnSettledIngestor { claimed: 0, enqueued: 0, unresolved: 0, + ignored: 0, invalid: 0, failed: 0, }; @@ -56,6 +78,18 @@ export class RuntimeTurnSettledIngestor { continue; } + const ignoredReason = getIgnoredReason(normalized.event); + if (ignoredReason) { + summary.ignored += 1; + await this.deps.eventStore.markProcessed(payload, { + event: normalized.event, + outcome: 'ignored', + reason: ignoredReason, + processedAt, + }); + continue; + } + const resolution = await this.deps.targetResolver.resolve(normalized.event); if (!resolution.ok) { summary.unresolved += 1; diff --git a/src/features/member-work-sync/core/application/RuntimeTurnSettledPorts.ts b/src/features/member-work-sync/core/application/RuntimeTurnSettledPorts.ts index ededdf62..5bccd83d 100644 --- a/src/features/member-work-sync/core/application/RuntimeTurnSettledPorts.ts +++ b/src/features/member-work-sync/core/application/RuntimeTurnSettledPorts.ts @@ -25,7 +25,7 @@ export interface RuntimeTurnSettledProcessedResult { event: RuntimeTurnSettledEvent; teamName?: string; memberName?: string; - outcome: 'enqueued' | 'unresolved' | 'duplicate'; + outcome: 'enqueued' | 'unresolved' | 'duplicate' | 'ignored'; reason?: string; processedAt: string; } diff --git a/src/main/services/team/opencode/bridge/OpenCodeReadinessBridge.ts b/src/main/services/team/opencode/bridge/OpenCodeReadinessBridge.ts index ad4dff69..373d3e9e 100644 --- a/src/main/services/team/opencode/bridge/OpenCodeReadinessBridge.ts +++ b/src/main/services/team/opencode/bridge/OpenCodeReadinessBridge.ts @@ -65,7 +65,9 @@ export interface OpenCodeReadinessBridgeCommandBody { const DEFAULT_READINESS_TIMEOUT_MS = 120_000; const DEFAULT_LAUNCH_TIMEOUT_MS = 120_000; const DEFAULT_RECONCILE_TIMEOUT_MS = 30_000; -const DEFAULT_SEND_TIMEOUT_MS = 30_000; +// Longer than the renderer-facing UI timeout: late OpenCode turns should still +// finish bridge-side observation and emit member-work-sync signals. +const DEFAULT_SEND_TIMEOUT_MS = 45_000; const DEFAULT_OBSERVE_TIMEOUT_MS = 8_000; const DEFAULT_STOP_TIMEOUT_MS = 30_000; const DEFAULT_CLEANUP_TIMEOUT_MS = 10_000; diff --git a/test/features/member-work-sync/main/RuntimeTurnSettledIngestor.test.ts b/test/features/member-work-sync/main/RuntimeTurnSettledIngestor.test.ts index 064799a2..d536ba8d 100644 --- a/test/features/member-work-sync/main/RuntimeTurnSettledIngestor.test.ts +++ b/test/features/member-work-sync/main/RuntimeTurnSettledIngestor.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from 'vitest'; import { RuntimeTurnSettledIngestor } from '@features/member-work-sync/core/application'; import { ClaudeStopHookPayloadNormalizer } from '@features/member-work-sync/main/infrastructure/ClaudeStopHookPayloadNormalizer'; import { NodeHashAdapter } from '@features/member-work-sync/main/infrastructure/NodeHashAdapter'; +import { OpenCodeTurnSettledPayloadNormalizer } from '@features/member-work-sync/main/infrastructure/OpenCodeTurnSettledPayloadNormalizer'; import type { RuntimeTurnSettledClaimedPayload, @@ -23,6 +24,17 @@ function makePayload(raw: string): RuntimeTurnSettledClaimedPayload { }; } +function makeOpenCodePayload(raw: string): RuntimeTurnSettledClaimedPayload { + return { + id: 'event-1.opencode.json', + fileName: 'event-1.opencode.json', + filePath: '/tmp/event-1.opencode.json', + provider: 'opencode', + raw, + claimedAt: '2026-04-29T12:00:00.000Z', + }; +} + describe('RuntimeTurnSettledIngestor', () => { it('normalizes Claude Stop payloads and enqueues resolved member reconcile', async () => { const payload = makePayload( @@ -57,6 +69,7 @@ describe('RuntimeTurnSettledIngestor', () => { claimed: 1, enqueued: 1, unresolved: 0, + ignored: 0, invalid: 0, failed: 0, }); @@ -133,6 +146,7 @@ describe('RuntimeTurnSettledIngestor', () => { await expect(ingestor.drainPending()).resolves.toMatchObject({ claimed: 1, unresolved: 1, + ignored: 0, enqueued: 0, }); expect(processed[0]).toMatchObject({ @@ -140,4 +154,130 @@ describe('RuntimeTurnSettledIngestor', () => { reason: 'no_matching_member_session', }); }); + + it('normalizes OpenCode turn-settled payloads and enqueues provider-owned member reconcile', async () => { + const payload = makeOpenCodePayload( + JSON.stringify({ + schemaVersion: 1, + provider: 'opencode', + source: 'agent-teams-orchestrator-opencode', + eventName: 'runtime_turn_settled', + hookEventName: 'Stop', + sessionId: 'ses-opencode-1', + runtimePromptMessageId: 'msg_123', + laneId: 'secondary:opencode:jack', + memberName: 'jack', + teamName: 'team-a', + cwd: '/tmp/project', + outcome: 'success', + recordedAt: '2026-04-29T12:00:00.000Z', + }) + ); + const processed: RuntimeTurnSettledProcessedResult[] = []; + const store: RuntimeTurnSettledEventStorePort = { + claimPending: vi.fn(async () => [payload]), + markProcessed: vi.fn(async (_payload, result) => { + processed.push(result); + }), + markInvalid: vi.fn(), + }; + const resolver: RuntimeTurnSettledTargetResolverPort = { + resolve: vi.fn(async () => ({ ok: true as const, teamName: 'team-a', memberName: 'jack' })), + }; + const enqueueRuntimeTurnSettled = vi.fn(); + + const ingestor = new RuntimeTurnSettledIngestor({ + eventStore: store, + normalizer: new OpenCodeTurnSettledPayloadNormalizer(new NodeHashAdapter()), + targetResolver: resolver, + reconcileQueue: { enqueueRuntimeTurnSettled }, + clock: { now: () => new Date('2026-04-29T12:01:00.000Z') }, + }); + + await expect(ingestor.drainPending()).resolves.toMatchObject({ + claimed: 1, + enqueued: 1, + invalid: 0, + unresolved: 0, + ignored: 0, + }); + expect(enqueueRuntimeTurnSettled).toHaveBeenCalledWith({ + teamName: 'team-a', + memberName: 'jack', + event: expect.objectContaining({ + provider: 'opencode', + hookEventName: 'Stop', + sessionId: 'ses-opencode-1', + turnId: 'msg_123', + teamName: 'team-a', + memberName: 'jack', + agentId: 'secondary:opencode:jack', + outcome: 'success', + }), + }); + expect(processed[0]).toMatchObject({ + outcome: 'enqueued', + teamName: 'team-a', + memberName: 'jack', + }); + }); + + it.each([ + 'timeout', + 'stream_unavailable', + 'prompt_rejected', + 'idle_without_assistant_activity', + 'unknown', + ])('ignores non-terminal OpenCode outcome %s without enqueueing premature reconcile', async (outcome) => { + const payload = makeOpenCodePayload( + JSON.stringify({ + schemaVersion: 1, + provider: 'opencode', + source: 'agent-teams-orchestrator-opencode', + eventName: 'runtime_turn_settled', + hookEventName: 'Stop', + sessionId: 'ses-opencode-1', + runtimePromptMessageId: 'msg_123', + laneId: 'secondary:opencode:jack', + memberName: 'jack', + teamName: 'team-a', + outcome, + recordedAt: '2026-04-29T12:00:00.000Z', + }) + ); + const processed: RuntimeTurnSettledProcessedResult[] = []; + const store: RuntimeTurnSettledEventStorePort = { + claimPending: vi.fn(async () => [payload]), + markProcessed: vi.fn(async (_payload, result) => { + processed.push(result); + }), + markInvalid: vi.fn(), + }; + const resolver: RuntimeTurnSettledTargetResolverPort = { + resolve: vi.fn(), + }; + const enqueueRuntimeTurnSettled = vi.fn(); + + const ingestor = new RuntimeTurnSettledIngestor({ + eventStore: store, + normalizer: new OpenCodeTurnSettledPayloadNormalizer(new NodeHashAdapter()), + targetResolver: resolver, + reconcileQueue: { enqueueRuntimeTurnSettled }, + clock: { now: () => new Date('2026-04-29T12:01:00.000Z') }, + }); + + await expect(ingestor.drainPending()).resolves.toMatchObject({ + claimed: 1, + enqueued: 0, + unresolved: 0, + ignored: 1, + invalid: 0, + }); + expect(resolver.resolve).not.toHaveBeenCalled(); + expect(enqueueRuntimeTurnSettled).not.toHaveBeenCalled(); + expect(processed[0]).toMatchObject({ + outcome: 'ignored', + reason: `opencode_non_terminal_outcome:${outcome}`, + }); + }); });