feat(member-work-sync): ingest codex turn settled events

This commit is contained in:
777genius 2026-04-29 20:43:29 +03:00
parent adb2b5c351
commit 8e7f7bea43
12 changed files with 424 additions and 1 deletions

View file

@ -11,6 +11,11 @@ export interface RuntimeTurnSettledEvent {
turnId?: string;
transcriptPath?: string;
cwd?: string;
teamName?: string;
memberName?: string;
agentId?: string;
threadId?: string;
outcome?: string;
}
export function buildRuntimeTurnSettledSourceId(input: {

View file

@ -81,6 +81,10 @@ export class TeamRuntimeTurnSettledTargetResolver
}
async resolve(event: RuntimeTurnSettledEvent): Promise<RuntimeTurnSettledTargetResolution> {
if (event.provider === 'codex') {
return this.resolveCodexEvent(event);
}
if (event.provider !== 'claude') {
return { ok: false, reason: 'unsupported_provider' };
}
@ -152,6 +156,35 @@ export class TeamRuntimeTurnSettledTargetResolver
};
}
private async resolveCodexEvent(
event: RuntimeTurnSettledEvent
): Promise<RuntimeTurnSettledTargetResolution> {
const teamName = event.teamName?.trim();
const memberName = event.memberName?.trim();
if (!teamName || !memberName) {
return { ok: false, reason: 'missing_team_member_identity' };
}
const member = await this.resolveActiveMember(teamName, memberName);
if (!member) {
return { ok: false, reason: 'member_not_active' };
}
if (isReservedMemberName(member.name)) {
return { ok: false, reason: 'reserved_member' };
}
const providerId = providerForMember(member);
if (providerId && providerId !== 'codex') {
return { ok: false, reason: 'provider_mismatch' };
}
return {
ok: true,
teamName,
memberName: normalizeMemberName(member.name),
};
}
private async resolveActiveMember(
teamName: string,
memberName: string

View file

@ -26,6 +26,8 @@ import { TeamTaskStallJournalWorkSyncCooldown } from '../adapters/output/TeamTas
import { TeamTaskAgendaSource } from '../adapters/output/TeamTaskAgendaSource';
import { TeamRuntimeTurnSettledTargetResolver } from '../adapters/output/TeamRuntimeTurnSettledTargetResolver';
import { ClaudeStopHookPayloadNormalizer } from '../infrastructure/ClaudeStopHookPayloadNormalizer';
import { CodexNativeTurnSettledPayloadNormalizer } from '../infrastructure/CodexNativeTurnSettledPayloadNormalizer';
import { CompositeRuntimeTurnSettledPayloadNormalizer } from '../infrastructure/CompositeRuntimeTurnSettledPayloadNormalizer';
import { FileRuntimeTurnSettledEventStore } from '../infrastructure/FileRuntimeTurnSettledEventStore';
import { HmacMemberWorkSyncReportTokenAdapter } from '../infrastructure/HmacMemberWorkSyncReportTokenAdapter';
import {
@ -40,6 +42,7 @@ import { NodeHashAdapter } from '../infrastructure/NodeHashAdapter';
import { RuntimeTurnSettledDrainScheduler } from '../infrastructure/RuntimeTurnSettledDrainScheduler';
import { RuntimeTurnSettledSpoolPaths } from '../infrastructure/RuntimeTurnSettledSpoolPaths';
import { ShellRuntimeTurnSettledHookScriptInstaller } from '../infrastructure/ShellRuntimeTurnSettledHookScriptInstaller';
import { buildRuntimeTurnSettledEnvironment } from '../infrastructure/runtimeTurnSettledEnvironment';
import { buildRuntimeTurnSettledHookSettings } from '../infrastructure/runtimeTurnSettledHookSettings';
import { SystemClockAdapter } from '../infrastructure/SystemClockAdapter';
@ -90,6 +93,9 @@ export interface MemberWorkSyncFeatureFacade {
buildRuntimeTurnSettledHookSettings(input: {
provider: RuntimeTurnSettledProvider;
}): Promise<Record<string, unknown> | null>;
buildRuntimeTurnSettledEnvironment(input: {
provider: RuntimeTurnSettledProvider;
}): Promise<Record<string, string> | null>;
drainRuntimeTurnSettledEvents(): Promise<RuntimeTurnSettledDrainSummary>;
getQueueDiagnostics(): MemberWorkSyncQueueDiagnostics;
dispose(): Promise<void>;
@ -127,7 +133,10 @@ export function createMemberWorkSyncFeature(deps: {
const runtimeTurnSettledStore = new FileRuntimeTurnSettledEventStore({
paths: runtimeTurnSettledSpoolPaths,
});
const runtimeTurnSettledNormalizer = new ClaudeStopHookPayloadNormalizer(hash);
const runtimeTurnSettledNormalizer = new CompositeRuntimeTurnSettledPayloadNormalizer([
new ClaudeStopHookPayloadNormalizer(hash),
new CodexNativeTurnSettledPayloadNormalizer(hash),
]);
const runtimeTurnSettledTargetResolver =
deps.runtimeTurnSettledTargetResolver ??
new TeamRuntimeTurnSettledTargetResolver({
@ -258,6 +267,16 @@ export function createMemberWorkSyncFeature(deps: {
provider,
});
},
buildRuntimeTurnSettledEnvironment: async ({ provider }) => {
if (provider !== 'codex') {
return null;
}
const installed = await runtimeTurnSettledHookInstaller.install();
return buildRuntimeTurnSettledEnvironment({
provider,
spoolRoot: installed.spoolRoot,
});
},
drainRuntimeTurnSettledEvents: () => runtimeTurnSettledIngestor.drainPending(),
getQueueDiagnostics: () => queue.getDiagnostics(),
dispose: async () => {

View file

@ -0,0 +1,113 @@
import {
buildRuntimeTurnSettledSourceId,
type RuntimeTurnSettledProvider,
} from '../../core/domain';
import type {
MemberWorkSyncHashPort,
RuntimeTurnSettledPayloadNormalization,
RuntimeTurnSettledPayloadNormalizerPort,
} from '../../core/application';
function asRecord(value: unknown): Record<string, unknown> | null {
return value && typeof value === 'object' && !Array.isArray(value)
? (value as Record<string, unknown>)
: null;
}
function getString(record: Record<string, unknown>, ...keys: string[]): string | undefined {
for (const key of keys) {
const value = record[key];
if (typeof value !== 'string') {
continue;
}
const trimmed = value.trim();
if (trimmed.length > 0) {
return trimmed;
}
}
return undefined;
}
export class CodexNativeTurnSettledPayloadNormalizer
implements RuntimeTurnSettledPayloadNormalizerPort
{
constructor(private readonly hash: MemberWorkSyncHashPort) {}
normalize(input: {
provider: RuntimeTurnSettledProvider;
raw: string;
recordedAt: string;
}): RuntimeTurnSettledPayloadNormalization {
if (input.provider !== 'codex') {
return { ok: false, reason: 'unsupported_provider' };
}
let parsed: unknown;
try {
parsed = JSON.parse(input.raw);
} catch {
return { ok: false, reason: 'invalid_json' };
}
const payload = asRecord(parsed);
if (!payload) {
return { ok: false, reason: 'payload_not_object' };
}
const provider = getString(payload, 'provider');
if (provider !== 'codex') {
return { ok: false, reason: 'provider_mismatch' };
}
const source = getString(payload, 'source');
if (source !== 'agent-teams-orchestrator-codex-native') {
return { ok: false, reason: 'source_mismatch' };
}
const eventName = getString(payload, 'eventName', 'event_name');
const hookEventName = getString(payload, 'hookEventName', 'hook_event_name');
if (eventName !== 'runtime_turn_settled' && hookEventName !== 'Stop') {
return { ok: false, reason: 'not_turn_settled_event' };
}
const sessionId = getString(payload, 'sessionId', 'session_id');
const teamName = getString(payload, 'teamName', 'team_name');
const memberName = getString(payload, 'memberName', 'member_name', 'agentName', 'agent_name');
if (!sessionId) {
return { ok: false, reason: 'missing_session_identity' };
}
if (!teamName || !memberName) {
return { ok: false, reason: 'missing_team_member_identity' };
}
const payloadHash = this.hash.sha256Hex(input.raw);
const threadId = getString(payload, 'threadId', 'thread_id');
const turnId = getString(payload, 'turnId', 'turn_id') ?? threadId;
const cwd = getString(payload, 'cwd');
const agentId = getString(payload, 'agentId', 'agent_id');
const outcome = getString(payload, 'outcome');
return {
ok: true,
event: {
schemaVersion: 1,
provider: 'codex',
hookEventName: 'Stop',
payloadHash,
recordedAt: getString(payload, 'recordedAt', 'recorded_at') ?? input.recordedAt,
sourceId: buildRuntimeTurnSettledSourceId({
provider: 'codex',
sessionId,
turnId,
payloadHash,
}),
sessionId,
...(turnId ? { turnId } : {}),
...(cwd ? { cwd } : {}),
teamName,
memberName,
...(agentId ? { agentId } : {}),
...(threadId ? { threadId } : {}),
...(outcome ? { outcome } : {}),
},
};
}
}

View file

@ -0,0 +1,32 @@
import type {
RuntimeTurnSettledPayloadNormalization,
RuntimeTurnSettledPayloadNormalizerPort,
} from '../../core/application';
import type { RuntimeTurnSettledProvider } from '../../core/domain';
export class CompositeRuntimeTurnSettledPayloadNormalizer
implements RuntimeTurnSettledPayloadNormalizerPort
{
constructor(
private readonly normalizers: readonly RuntimeTurnSettledPayloadNormalizerPort[]
) {}
normalize(input: {
provider: RuntimeTurnSettledProvider;
raw: string;
recordedAt: string;
}): RuntimeTurnSettledPayloadNormalization {
let lastUnsupportedReason = 'unsupported_provider';
for (const normalizer of this.normalizers) {
const result = normalizer.normalize(input);
if (result.ok) {
return result;
}
if (result.reason !== 'unsupported_provider') {
return result;
}
lastUnsupportedReason = result.reason;
}
return { ok: false, reason: lastUnsupportedReason };
}
}

View file

@ -0,0 +1,16 @@
import type { RuntimeTurnSettledProvider } from '../../core/domain';
export const RUNTIME_TURN_SETTLED_SPOOL_ROOT_ENV =
'AGENT_TEAMS_RUNTIME_TURN_SETTLED_SPOOL_ROOT';
export function buildRuntimeTurnSettledEnvironment(input: {
provider: RuntimeTurnSettledProvider;
spoolRoot: string;
}): Record<string, string> | null {
if (input.provider !== 'codex') {
return null;
}
return {
[RUNTIME_TURN_SETTLED_SPOOL_ROOT_ENV]: input.spoolRoot,
};
}

View file

@ -1254,6 +1254,11 @@ async function initializeServices(): Promise<void> {
? memberWorkSyncFeature.buildRuntimeTurnSettledHookSettings(input)
: Promise.resolve(null)
);
teamProvisioningService.setRuntimeTurnSettledEnvironmentProvider((input) =>
memberWorkSyncFeature
? memberWorkSyncFeature.buildRuntimeTurnSettledEnvironment(input)
: Promise.resolve(null)
);
void teamDataService
.listTeams()
.then(async (teams) => {

View file

@ -4324,6 +4324,9 @@ export class TeamProvisioningService {
private runtimeTurnSettledHookSettingsProvider:
| ((input: { provider: RuntimeTurnSettledProvider }) => Promise<Record<string, unknown> | null>)
| null = null;
private runtimeTurnSettledEnvironmentProvider:
| ((input: { provider: RuntimeTurnSettledProvider }) => Promise<Record<string, string> | null>)
| null = null;
private readonly stoppedTeamOpenCodeRuntimeCleanupInFlight = new Map<string, Promise<number>>();
private readonly cleanedStoppedTeamOpenCodeRuntimeLanes = new Set<string>();
private crossTeamSender:
@ -4395,6 +4398,16 @@ export class TeamProvisioningService {
this.runtimeTurnSettledHookSettingsProvider = provider;
}
setRuntimeTurnSettledEnvironmentProvider(
provider:
| ((input: {
provider: RuntimeTurnSettledProvider;
}) => Promise<Record<string, string> | null>)
| null
): void {
this.runtimeTurnSettledEnvironmentProvider = provider;
}
private async buildRuntimeTurnSettledHookSettingsArgs(
providerId: TeamProviderId
): Promise<string[]> {
@ -4415,6 +4428,25 @@ export class TeamProvisioningService {
}
}
private async buildRuntimeTurnSettledEnvironment(
providerId: TeamProviderId
): Promise<Record<string, string>> {
if (providerId !== 'codex' || !this.runtimeTurnSettledEnvironmentProvider) {
return {};
}
try {
return (await this.runtimeTurnSettledEnvironmentProvider({ provider: 'codex' })) ?? {};
} catch (error) {
logger.warn(
`Failed to build member work sync runtime turn-settled environment: ${
error instanceof Error ? error.message : String(error)
}`
);
return {};
}
}
private async readRuntimeProviderLaunchFacts(params: {
claudePath: string;
cwd: string;
@ -22816,6 +22848,7 @@ export class TeamProvisioningService {
});
const providerConnectionIssue = providerEnvResult.connectionIssues[resolvedProviderId];
const providerEnv = providerEnvResult.env;
Object.assign(providerEnv, await this.buildRuntimeTurnSettledEnvironment(resolvedProviderId));
const controlApiBaseUrl = await this.resolveControlApiBaseUrl();
if (controlApiBaseUrl) {

View file

@ -0,0 +1,64 @@
import { describe, expect, it } from 'vitest';
import { CodexNativeTurnSettledPayloadNormalizer } from '@features/member-work-sync/main/infrastructure/CodexNativeTurnSettledPayloadNormalizer';
import { NodeHashAdapter } from '@features/member-work-sync/main/infrastructure/NodeHashAdapter';
describe('CodexNativeTurnSettledPayloadNormalizer', () => {
it('normalizes orchestrator-native Codex turn-settled payloads', () => {
const normalizer = new CodexNativeTurnSettledPayloadNormalizer(new NodeHashAdapter());
const result = normalizer.normalize({
provider: 'codex',
raw: JSON.stringify({
schemaVersion: 1,
provider: 'codex',
source: 'agent-teams-orchestrator-codex-native',
eventName: 'runtime_turn_settled',
hookEventName: 'Stop',
sessionId: 'ses-1',
threadId: 'thread-1',
agentId: 'jack@team-a',
agentName: 'jack',
teamName: 'team-a',
cwd: '/tmp/project',
outcome: 'success',
recordedAt: '2026-04-29T12:00:00.000Z',
}),
recordedAt: '2026-04-29T12:00:01.000Z',
});
expect(result).toEqual({
ok: true,
event: expect.objectContaining({
provider: 'codex',
hookEventName: 'Stop',
sessionId: 'ses-1',
turnId: 'thread-1',
threadId: 'thread-1',
teamName: 'team-a',
memberName: 'jack',
agentId: 'jack@team-a',
cwd: '/tmp/project',
outcome: 'success',
recordedAt: '2026-04-29T12:00:00.000Z',
}),
});
});
it('rejects Codex payloads without durable team/member identity', () => {
const normalizer = new CodexNativeTurnSettledPayloadNormalizer(new NodeHashAdapter());
expect(
normalizer.normalize({
provider: 'codex',
raw: JSON.stringify({
provider: 'codex',
source: 'agent-teams-orchestrator-codex-native',
eventName: 'runtime_turn_settled',
sessionId: 'ses-1',
}),
recordedAt: '2026-04-29T12:00:01.000Z',
})
).toEqual({ ok: false, reason: 'missing_team_member_identity' });
});
});

View file

@ -74,4 +74,63 @@ describe('TeamRuntimeTurnSettledTargetResolver', () => {
})
).resolves.toEqual({ ok: false, reason: 'provider_mismatch' });
});
it('resolves Codex native turn-settled payloads from durable team/member identity', async () => {
const resolver = new TeamRuntimeTurnSettledTargetResolver({
teamSource: {
listTeams: vi.fn(async () => {
throw new Error('codex path should not scan attributed files');
}),
getConfig: vi.fn(async () => ({
name: 'team-a',
members: [{ name: 'Jack', providerId: 'codex' }],
}) satisfies TeamConfig),
},
membersMetaStore: { getMembers: vi.fn(async () => []) } as never,
memberLogsFinder: {
listAttributedMemberFiles: vi.fn(async () => []),
},
});
await expect(
resolver.resolve({
schemaVersion: 1,
provider: 'codex',
hookEventName: 'Stop',
sourceId: 'source-1',
payloadHash: 'hash',
recordedAt: '2026-04-29T12:00:00.000Z',
sessionId: 'ses-1',
teamName: 'team-a',
memberName: 'jack',
})
).resolves.toEqual({ ok: true, teamName: 'team-a', memberName: 'jack' });
});
it('rejects Codex native events for non-Codex teammates', async () => {
const resolver = new TeamRuntimeTurnSettledTargetResolver({
teamSource: {
listTeams: vi.fn(async () => []),
getConfig: vi.fn(async () => ({
name: 'team-a',
members: [{ name: 'Jack', providerId: 'anthropic' }],
}) satisfies TeamConfig),
},
membersMetaStore: { getMembers: vi.fn(async () => []) } as never,
});
await expect(
resolver.resolve({
schemaVersion: 1,
provider: 'codex',
hookEventName: 'Stop',
sourceId: 'source-1',
payloadHash: 'hash',
recordedAt: '2026-04-29T12:00:00.000Z',
sessionId: 'ses-1',
teamName: 'team-a',
memberName: 'jack',
})
).resolves.toEqual({ ok: false, reason: 'provider_mismatch' });
});
});

View file

@ -8,6 +8,7 @@ import {
createMemberWorkSyncFeature,
resolveMemberWorkSyncNudgeSideEffectsEnabled,
} from '@features/member-work-sync/main';
import { RUNTIME_TURN_SETTLED_SPOOL_ROOT_ENV } from '@features/member-work-sync/main/infrastructure/runtimeTurnSettledEnvironment';
const tempRoots: string[] = [];
@ -114,4 +115,31 @@ describe('createMemberWorkSyncFeature composition', () => {
await feature.dispose();
}
});
it('builds Codex turn-settled environment without requiring nudge side effects', async () => {
const root = makeTempRoot();
const feature = createMemberWorkSyncFeature({
teamsBasePath: root,
configReader: {} as never,
taskReader: {} as never,
kanbanManager: {} as never,
membersMetaStore: {} as never,
nudgeSideEffectsEnabled: false,
});
try {
const env = await feature.buildRuntimeTurnSettledEnvironment({ provider: 'codex' });
expect(env).toEqual({
[RUNTIME_TURN_SETTLED_SPOOL_ROOT_ENV]: path.join(
root,
'.member-work-sync/runtime-hooks'
),
});
await expect(
fs.promises.stat(path.join(root, '.member-work-sync/runtime-hooks/incoming'))
).resolves.toMatchObject({ mode: expect.any(Number) });
} finally {
await feature.dispose();
}
});
});

View file

@ -149,6 +149,8 @@ const REQUIRED_MOCK_AGENT_TEAMS_TOOLS = [
'lead_briefing',
'member_briefing',
'message_send',
'member_work_sync_report',
'member_work_sync_status',
'process_list',
'process_register',
'process_stop',
@ -1954,6 +1956,20 @@ describe('TeamProvisioningService prepare/auth behavior', () => {
expect(result.env.ANTHROPIC_API_KEY).toBe('real-key');
});
it('adds member-work-sync turn-settled spool env for Codex provisioning', async () => {
const svc = new TeamProvisioningService();
svc.setRuntimeTurnSettledEnvironmentProvider(async ({ provider }) =>
provider === 'codex'
? { AGENT_TEAMS_RUNTIME_TURN_SETTLED_SPOOL_ROOT: '/tmp/runtime-hooks' }
: null
);
const result = await (svc as any).buildProvisioningEnv('codex');
expect(result.authSource).toBe('codex_runtime');
expect(result.env.AGENT_TEAMS_RUNTIME_TURN_SETTLED_SPOOL_ROOT).toBe('/tmp/runtime-hooks');
});
it('allows help-env resolution to continue even when provisioning env warns', async () => {
const svc = new TeamProvisioningService();
vi.spyOn(svc as any, 'buildProvisioningEnv').mockResolvedValue({