fix(team): load opencode task logs from delivery session evidence
This commit is contained in:
777genius
parent
662691b24b
commit
dd412355d2
5 changed files with 680 additions and 12 deletions
|
|
@ -13,12 +13,17 @@ const logger = createLogger('OpenCodeTaskLogAttributionStore');
|
|||
const MAX_ATTRIBUTION_FILE_BYTES = 512 * 1024;
|
||||
|
||||
export type OpenCodeTaskLogAttributionScope = 'task_session' | 'member_session_window';
|
||||
export type OpenCodeTaskLogAttributionSource = 'manual' | 'launch_runtime' | 'reconcile';
|
||||
export type OpenCodeTaskLogAttributionSource =
|
||||
| 'manual'
|
||||
| 'launch_runtime'
|
||||
| 'reconcile'
|
||||
| 'delivery_ledger';
|
||||
|
||||
export interface OpenCodeTaskLogAttributionRecord {
|
||||
taskId: string;
|
||||
memberName: string;
|
||||
scope: OpenCodeTaskLogAttributionScope;
|
||||
laneId?: string;
|
||||
sessionId?: string;
|
||||
since?: string;
|
||||
until?: string;
|
||||
|
|
@ -66,7 +71,10 @@ function normalizeScope(value: unknown): OpenCodeTaskLogAttributionScope {
|
|||
}
|
||||
|
||||
function normalizeSource(value: unknown): OpenCodeTaskLogAttributionSource | undefined {
|
||||
return value === 'manual' || value === 'launch_runtime' || value === 'reconcile'
|
||||
return value === 'manual' ||
|
||||
value === 'launch_runtime' ||
|
||||
value === 'reconcile' ||
|
||||
value === 'delivery_ledger'
|
||||
? value
|
||||
: undefined;
|
||||
}
|
||||
|
|
@ -86,6 +94,7 @@ function normalizeRecord(
|
|||
return null;
|
||||
}
|
||||
const sessionId = trimString(raw.sessionId);
|
||||
const laneId = trimString(raw.laneId);
|
||||
const startMessageUuid = trimString(raw.startMessageUuid);
|
||||
const endMessageUuid = trimString(raw.endMessageUuid);
|
||||
const source = normalizeSource(raw.source);
|
||||
|
|
@ -96,6 +105,7 @@ function normalizeRecord(
|
|||
taskId,
|
||||
memberName,
|
||||
scope: normalizeScope(raw.scope),
|
||||
...(laneId ? { laneId } : {}),
|
||||
...(sessionId ? { sessionId } : {}),
|
||||
...(since ? { since } : {}),
|
||||
...(until ? { until } : {}),
|
||||
|
|
|
|||
|
|
@ -4,16 +4,19 @@ import { ClaudeMultimodelBridgeService } from '../../../runtime/ClaudeMultimodel
|
|||
import { canonicalizeAgentTeamsToolName } from '../../agentTeamsToolNames';
|
||||
import { ClaudeBinaryResolver } from '../../ClaudeBinaryResolver';
|
||||
import { TeamTaskReader } from '../../TeamTaskReader';
|
||||
import { getTeamsBasePath } from '@main/utils/pathDecoder';
|
||||
import { BoardTaskExactLogChunkBuilder } from '../exact/BoardTaskExactLogChunkBuilder';
|
||||
|
||||
import { mapOpenCodeRuntimeTranscriptLogMessageToParsedMessage } from './OpenCodeRuntimeProjectionMapper';
|
||||
import { OpenCodeTaskLogAttributionStore } from './OpenCodeTaskLogAttributionStore';
|
||||
import { TaskLogOpenCodeSessionEvidenceSource } from './TaskLogOpenCodeSessionEvidenceSource';
|
||||
|
||||
import type { OpenCodeRuntimeTranscriptLogMessage } from '../../../runtime/ClaudeMultimodelBridgeService';
|
||||
import type {
|
||||
OpenCodeTaskLogAttributionReader,
|
||||
OpenCodeTaskLogAttributionRecord,
|
||||
} from './OpenCodeTaskLogAttributionStore';
|
||||
import type { OpenCodeTaskLogSessionEvidenceReader } from './TaskLogOpenCodeSessionEvidenceSource';
|
||||
import type { ParsedMessage } from '@main/types';
|
||||
import type {
|
||||
BoardTaskLogActor,
|
||||
|
|
@ -191,6 +194,7 @@ function stableAttributionKey(records: OpenCodeTaskLogAttributionRecord[]): stri
|
|||
JSON.stringify([
|
||||
normalizeMemberName(record.memberName),
|
||||
record.scope,
|
||||
record.laneId ?? '',
|
||||
record.sessionId ?? '',
|
||||
record.since ?? '',
|
||||
record.until ?? '',
|
||||
|
|
@ -202,6 +206,34 @@ function stableAttributionKey(records: OpenCodeTaskLogAttributionRecord[]): stri
|
|||
.join('|');
|
||||
}
|
||||
|
||||
function mergeTaskLogAttributionRecords(
|
||||
attributionRecords: OpenCodeTaskLogAttributionRecord[],
|
||||
sessionEvidenceRecords: OpenCodeTaskLogAttributionRecord[]
|
||||
): OpenCodeTaskLogAttributionRecord[] {
|
||||
const merged: OpenCodeTaskLogAttributionRecord[] = [];
|
||||
const seen = new Set<string>();
|
||||
for (const record of [...attributionRecords, ...sessionEvidenceRecords]) {
|
||||
const key = JSON.stringify([
|
||||
record.taskId,
|
||||
normalizeMemberName(record.memberName),
|
||||
record.scope,
|
||||
record.laneId ?? '',
|
||||
record.sessionId ?? '',
|
||||
record.since ?? '',
|
||||
record.until ?? '',
|
||||
record.startMessageUuid ?? '',
|
||||
record.endMessageUuid ?? '',
|
||||
record.source ?? '',
|
||||
]);
|
||||
if (seen.has(key)) {
|
||||
continue;
|
||||
}
|
||||
seen.add(key);
|
||||
merged.push(record);
|
||||
}
|
||||
return merged;
|
||||
}
|
||||
|
||||
function normalizeTaskRef(value: unknown): string | null {
|
||||
if (typeof value !== 'string' && typeof value !== 'number') {
|
||||
return null;
|
||||
|
|
@ -959,7 +991,12 @@ export class OpenCodeTaskLogStreamSource {
|
|||
private readonly binaryResolver: BinaryResolverLike = ClaudeBinaryResolver,
|
||||
private readonly taskReader: TeamTaskReader = new TeamTaskReader(),
|
||||
private readonly chunkBuilder: BoardTaskExactLogChunkBuilder = new BoardTaskExactLogChunkBuilder(),
|
||||
private readonly attributionStore: OpenCodeTaskLogAttributionReader = new OpenCodeTaskLogAttributionStore()
|
||||
private readonly attributionStore: OpenCodeTaskLogAttributionReader = new OpenCodeTaskLogAttributionStore(),
|
||||
private readonly sessionEvidenceSource: OpenCodeTaskLogSessionEvidenceReader = new TaskLogOpenCodeSessionEvidenceSource(
|
||||
{
|
||||
teamsBasePath: getTeamsBasePath(),
|
||||
}
|
||||
)
|
||||
) {}
|
||||
|
||||
private async resolveTask(teamName: string, taskId: string): Promise<TeamTask | null> {
|
||||
|
|
@ -979,12 +1016,26 @@ export class OpenCodeTaskLogStreamSource {
|
|||
return null;
|
||||
}
|
||||
|
||||
const attributionRecords = await this.attributionStore.readTaskRecords(teamName, taskId);
|
||||
if (!task.owner?.trim() && attributionRecords.length === 0) {
|
||||
const [attributionRecords, sessionEvidenceRecords] = await Promise.all([
|
||||
this.attributionStore.readTaskRecords(teamName, taskId),
|
||||
this.sessionEvidenceSource.readTaskRecords(teamName, task).catch((error) => {
|
||||
logger.warn(
|
||||
`[${teamName}/${task.id}] OpenCode task-log session evidence lookup failed: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`
|
||||
);
|
||||
return [];
|
||||
}),
|
||||
]);
|
||||
const taskLogRecords = mergeTaskLogAttributionRecords(
|
||||
attributionRecords,
|
||||
sessionEvidenceRecords
|
||||
);
|
||||
if (!task.owner?.trim() && taskLogRecords.length === 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const cacheKey = `${teamName}::${stableTaskWindowKey(task)}::${stableAttributionKey(attributionRecords)}`;
|
||||
const cacheKey = `${teamName}::${stableTaskWindowKey(task)}::${stableAttributionKey(taskLogRecords)}`;
|
||||
const cached = this.cache.get(cacheKey);
|
||||
if (cached && cached.expiresAt > Date.now()) {
|
||||
return cached.response;
|
||||
|
|
@ -995,7 +1046,7 @@ export class OpenCodeTaskLogStreamSource {
|
|||
return await existingPromise;
|
||||
}
|
||||
|
||||
const promise = this.buildTaskLogStream(teamName, task, attributionRecords)
|
||||
const promise = this.buildTaskLogStream(teamName, task, taskLogRecords)
|
||||
.catch((error) => {
|
||||
logger.warn(
|
||||
`[${teamName}/${task.id}] OpenCode task-log fallback failed: ${
|
||||
|
|
@ -1169,8 +1220,11 @@ export class OpenCodeTaskLogStreamSource {
|
|||
}
|
||||
|
||||
const memberKey = normalizeMemberName(memberName);
|
||||
const laneId = record.laneId?.trim();
|
||||
const sessionId = record.sessionId?.trim();
|
||||
const transcriptCacheKey = `${memberKey}::${sessionId ?? 'current'}`;
|
||||
const transcriptCacheKey = `${memberKey}::${laneId ?? 'current-lane'}::${
|
||||
sessionId ?? 'current'
|
||||
}`;
|
||||
if (!transcriptCache.has(transcriptCacheKey)) {
|
||||
transcriptCache.set(
|
||||
transcriptCacheKey,
|
||||
|
|
@ -1178,6 +1232,7 @@ export class OpenCodeTaskLogStreamSource {
|
|||
teamId: teamName,
|
||||
memberName,
|
||||
limit: ATTRIBUTED_TRANSCRIPT_LIMIT,
|
||||
...(laneId ? { laneId } : {}),
|
||||
...(sessionId ? { sessionId } : {}),
|
||||
})
|
||||
);
|
||||
|
|
@ -1191,10 +1246,13 @@ export class OpenCodeTaskLogStreamSource {
|
|||
continue;
|
||||
}
|
||||
|
||||
const filteredMessages = filterMessagesForAttribution(
|
||||
transcript.logProjection?.messages ?? [],
|
||||
record
|
||||
);
|
||||
const projectedMessages = transcript.logProjection?.messages ?? [];
|
||||
const markerProjection =
|
||||
record.source === 'delivery_ledger'
|
||||
? buildTaskMarkerProjection(projectedMessages, teamName, task)
|
||||
: null;
|
||||
const filteredMessages =
|
||||
markerProjection?.messages ?? filterMessagesForAttribution(projectedMessages, record);
|
||||
if (filteredMessages.length === 0) {
|
||||
continue;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,305 @@
|
|||
import { readdir } from 'node:fs/promises';
|
||||
import * as path from 'node:path';
|
||||
|
||||
import {
|
||||
createOpenCodePromptDeliveryLedgerStore,
|
||||
type OpenCodePromptDeliveryLedgerRecord,
|
||||
} from '../../opencode/delivery/OpenCodePromptDeliveryLedger';
|
||||
import {
|
||||
getOpenCodeLaneScopedRuntimeFilePath,
|
||||
getOpenCodeTeamRuntimeDirectory,
|
||||
readOpenCodeRuntimeLaneIndex,
|
||||
} from '../../opencode/store/OpenCodeRuntimeManifestEvidenceReader';
|
||||
|
||||
import type { OpenCodeTaskLogAttributionRecord } from './OpenCodeTaskLogAttributionStore';
|
||||
import type { TeamTask } from '@shared/types';
|
||||
|
||||
const OPENCODE_PROMPT_DELIVERY_LEDGER_FILE = 'opencode-prompt-delivery-ledger.json';
|
||||
const OPENCODE_TEAM_RUNTIME_LANES_DIR = 'lanes';
|
||||
const MAX_LEDGER_FILES_TO_SCAN = 48;
|
||||
const MAX_RECORDS_PER_LEDGER = 96;
|
||||
const MAX_EVIDENCE_RECORDS = 3;
|
||||
const TERMINAL_EVIDENCE_GRACE_MS = 5 * 60_000;
|
||||
|
||||
interface TaskLogOpenCodeSessionEvidenceSourceOptions {
|
||||
teamsBasePath: string;
|
||||
maxLedgerFilesToScan?: number;
|
||||
maxRecordsPerLedger?: number;
|
||||
maxEvidenceRecords?: number;
|
||||
}
|
||||
|
||||
export interface OpenCodeTaskLogSessionEvidenceReader {
|
||||
readTaskRecords(teamName: string, task: TeamTask): Promise<OpenCodeTaskLogAttributionRecord[]>;
|
||||
}
|
||||
|
||||
function normalizeTaskRef(value: unknown): string | null {
|
||||
if (typeof value !== 'string' && typeof value !== 'number') {
|
||||
return null;
|
||||
}
|
||||
const normalized = String(value).trim().replace(/^#/, '').toLowerCase();
|
||||
return normalized.length > 0 ? normalized : null;
|
||||
}
|
||||
|
||||
function buildTaskRefSet(task: TeamTask): Set<string> {
|
||||
return new Set(
|
||||
[task.id, task.displayId, task.sourceMessageId]
|
||||
.map(normalizeTaskRef)
|
||||
.filter((value): value is string => value !== null)
|
||||
);
|
||||
}
|
||||
|
||||
function parseTimestampMs(value: string | null | undefined): number {
|
||||
if (!value) {
|
||||
return 0;
|
||||
}
|
||||
const parsed = Date.parse(value);
|
||||
return Number.isFinite(parsed) ? parsed : 0;
|
||||
}
|
||||
|
||||
function minTimestampIso(values: Array<string | null | undefined>): string | undefined {
|
||||
const times = values.map(parseTimestampMs).filter((value) => Number.isFinite(value) && value > 0);
|
||||
if (times.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
return new Date(Math.min(...times)).toISOString();
|
||||
}
|
||||
|
||||
function maxTimestampIso(values: Array<string | null | undefined>): string | undefined {
|
||||
const times = values.map(parseTimestampMs).filter((value) => Number.isFinite(value) && value > 0);
|
||||
if (times.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
return new Date(Math.max(...times)).toISOString();
|
||||
}
|
||||
|
||||
function addMsToIso(value: string | undefined, deltaMs: number): string | undefined {
|
||||
if (!value) {
|
||||
return undefined;
|
||||
}
|
||||
const timestamp = Date.parse(value);
|
||||
if (!Number.isFinite(timestamp)) {
|
||||
return undefined;
|
||||
}
|
||||
return new Date(timestamp + deltaMs).toISOString();
|
||||
}
|
||||
|
||||
function recordReferencesTask(
|
||||
record: OpenCodePromptDeliveryLedgerRecord,
|
||||
taskRefs: Set<string>,
|
||||
task: TeamTask
|
||||
): boolean {
|
||||
if (task.sourceMessageId && record.inboxMessageId === task.sourceMessageId) {
|
||||
return true;
|
||||
}
|
||||
return record.taskRefs.some((ref) => {
|
||||
const taskId = normalizeTaskRef(ref.taskId);
|
||||
const displayId = normalizeTaskRef(ref.displayId);
|
||||
return Boolean((taskId && taskRefs.has(taskId)) || (displayId && taskRefs.has(displayId)));
|
||||
});
|
||||
}
|
||||
|
||||
function isTerminalTask(task: TeamTask): boolean {
|
||||
return task.status === 'completed' || task.status === 'pending' || task.status === 'deleted';
|
||||
}
|
||||
|
||||
function shouldUseRecord(
|
||||
record: OpenCodePromptDeliveryLedgerRecord,
|
||||
teamName: string,
|
||||
task: TeamTask,
|
||||
taskRefs: Set<string>
|
||||
): boolean {
|
||||
return (
|
||||
record.teamName === teamName &&
|
||||
Boolean(record.runtimeSessionId?.trim()) &&
|
||||
!(record.status === 'failed_terminal' && !record.acceptedAt) &&
|
||||
recordReferencesTask(record, taskRefs, task)
|
||||
);
|
||||
}
|
||||
|
||||
function recordSortTimestamp(record: OpenCodePromptDeliveryLedgerRecord): number {
|
||||
return Math.max(
|
||||
parseTimestampMs(record.respondedAt),
|
||||
parseTimestampMs(record.lastObservedAt),
|
||||
parseTimestampMs(record.acceptedAt),
|
||||
parseTimestampMs(record.lastAttemptAt),
|
||||
parseTimestampMs(record.inboxTimestamp),
|
||||
parseTimestampMs(record.updatedAt),
|
||||
parseTimestampMs(record.createdAt),
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
function toAttributionRecord(
|
||||
record: OpenCodePromptDeliveryLedgerRecord,
|
||||
task: TeamTask
|
||||
): OpenCodeTaskLogAttributionRecord | null {
|
||||
const sessionId = record.runtimeSessionId?.trim();
|
||||
const memberName = record.memberName.trim();
|
||||
if (!sessionId || !memberName) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const since = minTimestampIso([
|
||||
record.inboxTimestamp,
|
||||
record.acceptedAt,
|
||||
record.lastAttemptAt,
|
||||
record.createdAt,
|
||||
]);
|
||||
const terminalUntil = isTerminalTask(task)
|
||||
? maxTimestampIso([task.updatedAt, record.respondedAt, record.lastObservedAt, record.updatedAt])
|
||||
: undefined;
|
||||
const fallbackUntil =
|
||||
record.status === 'responded' || record.status === 'failed_terminal'
|
||||
? maxTimestampIso([
|
||||
record.respondedAt,
|
||||
record.lastObservedAt,
|
||||
record.failedAt,
|
||||
record.updatedAt,
|
||||
])
|
||||
: undefined;
|
||||
const until = addMsToIso(terminalUntil ?? fallbackUntil, TERMINAL_EVIDENCE_GRACE_MS);
|
||||
const startMessageUuid = record.deliveredUserMessageId?.trim() || undefined;
|
||||
|
||||
return {
|
||||
taskId: task.id,
|
||||
memberName,
|
||||
scope: 'member_session_window',
|
||||
laneId: record.laneId.trim(),
|
||||
sessionId,
|
||||
source: 'delivery_ledger',
|
||||
...(since ? { since } : {}),
|
||||
...(until ? { until } : {}),
|
||||
...(startMessageUuid ? { startMessageUuid } : {}),
|
||||
createdAt: record.createdAt,
|
||||
updatedAt: record.updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
async function mapWithConcurrency<TInput, TOutput>(
|
||||
inputs: readonly TInput[],
|
||||
concurrency: number,
|
||||
mapper: (input: TInput) => Promise<TOutput>
|
||||
): Promise<TOutput[]> {
|
||||
const results: TOutput[] = [];
|
||||
let index = 0;
|
||||
const workerCount = Math.max(1, Math.min(concurrency, inputs.length));
|
||||
await Promise.all(
|
||||
Array.from({ length: workerCount }, async () => {
|
||||
while (index < inputs.length) {
|
||||
const currentIndex = index;
|
||||
index += 1;
|
||||
results[currentIndex] = await mapper(inputs[currentIndex] as TInput);
|
||||
}
|
||||
})
|
||||
);
|
||||
return results;
|
||||
}
|
||||
|
||||
export class TaskLogOpenCodeSessionEvidenceSource implements OpenCodeTaskLogSessionEvidenceReader {
|
||||
private readonly teamsBasePath: string;
|
||||
private readonly maxLedgerFilesToScan: number;
|
||||
private readonly maxRecordsPerLedger: number;
|
||||
private readonly maxEvidenceRecords: number;
|
||||
|
||||
constructor(options: TaskLogOpenCodeSessionEvidenceSourceOptions) {
|
||||
this.teamsBasePath = options.teamsBasePath;
|
||||
this.maxLedgerFilesToScan = options.maxLedgerFilesToScan ?? MAX_LEDGER_FILES_TO_SCAN;
|
||||
this.maxRecordsPerLedger = options.maxRecordsPerLedger ?? MAX_RECORDS_PER_LEDGER;
|
||||
this.maxEvidenceRecords = options.maxEvidenceRecords ?? MAX_EVIDENCE_RECORDS;
|
||||
}
|
||||
|
||||
async readTaskRecords(
|
||||
teamName: string,
|
||||
task: TeamTask
|
||||
): Promise<OpenCodeTaskLogAttributionRecord[]> {
|
||||
const taskRefs = buildTaskRefSet(task);
|
||||
if (taskRefs.size === 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const ledgerPaths = await this.discoverLedgerPaths(teamName);
|
||||
if (ledgerPaths.length === 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const recordBatches = await mapWithConcurrency(ledgerPaths, 4, async (filePath) =>
|
||||
this.readLedgerRecords(filePath)
|
||||
);
|
||||
const records = recordBatches
|
||||
.flat()
|
||||
.filter((record) => shouldUseRecord(record, teamName, task, taskRefs))
|
||||
.sort((left, right) => recordSortTimestamp(right) - recordSortTimestamp(left));
|
||||
|
||||
const seen = new Set<string>();
|
||||
const result: OpenCodeTaskLogAttributionRecord[] = [];
|
||||
for (const record of records) {
|
||||
const sessionId = record.runtimeSessionId?.trim();
|
||||
if (!sessionId) {
|
||||
continue;
|
||||
}
|
||||
const key = [
|
||||
record.memberName.trim().toLowerCase(),
|
||||
record.laneId.trim(),
|
||||
sessionId,
|
||||
record.deliveredUserMessageId ?? record.inboxMessageId,
|
||||
].join('::');
|
||||
if (seen.has(key)) {
|
||||
continue;
|
||||
}
|
||||
seen.add(key);
|
||||
const attributionRecord = toAttributionRecord(record, task);
|
||||
if (!attributionRecord) {
|
||||
continue;
|
||||
}
|
||||
result.push(attributionRecord);
|
||||
if (result.length >= this.maxEvidenceRecords) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private async discoverLedgerPaths(teamName: string): Promise<string[]> {
|
||||
const ledgerPaths = new Set<string>();
|
||||
const runtimeDir = getOpenCodeTeamRuntimeDirectory(this.teamsBasePath, teamName);
|
||||
const lanesDir = path.join(runtimeDir, OPENCODE_TEAM_RUNTIME_LANES_DIR);
|
||||
const laneDirs = await readdir(lanesDir, { withFileTypes: true }).catch(() => []);
|
||||
for (const entry of laneDirs) {
|
||||
if (!entry.isDirectory()) {
|
||||
continue;
|
||||
}
|
||||
ledgerPaths.add(path.join(lanesDir, entry.name, OPENCODE_PROMPT_DELIVERY_LEDGER_FILE));
|
||||
if (ledgerPaths.size >= this.maxLedgerFilesToScan) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const laneIndex = await readOpenCodeRuntimeLaneIndex(this.teamsBasePath, teamName).catch(
|
||||
() => null
|
||||
);
|
||||
for (const laneId of Object.keys(laneIndex?.lanes ?? {})) {
|
||||
if (ledgerPaths.size >= this.maxLedgerFilesToScan) {
|
||||
break;
|
||||
}
|
||||
ledgerPaths.add(
|
||||
getOpenCodeLaneScopedRuntimeFilePath({
|
||||
teamsBasePath: this.teamsBasePath,
|
||||
teamName,
|
||||
laneId,
|
||||
fileName: OPENCODE_PROMPT_DELIVERY_LEDGER_FILE,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
return Array.from(ledgerPaths);
|
||||
}
|
||||
|
||||
private async readLedgerRecords(filePath: string): Promise<OpenCodePromptDeliveryLedgerRecord[]> {
|
||||
const store = createOpenCodePromptDeliveryLedgerStore({ filePath });
|
||||
return await store
|
||||
.list()
|
||||
.then((records) => records.slice(-this.maxRecordsPerLedger))
|
||||
.catch(() => []);
|
||||
}
|
||||
}
|
||||
|
|
@ -264,6 +264,95 @@ describe('OpenCodeTaskLogStreamSource', () => {
|
|||
expect(second).toEqual(first);
|
||||
});
|
||||
|
||||
it('uses exact OpenCode session evidence from delivery ledgers before current-lane fallback', async () => {
|
||||
const bridge = {
|
||||
getOpenCodeTranscript: vi.fn(async (_binaryPath, request: { sessionId?: string }) => ({
|
||||
sessionId: request.sessionId ?? 'current-session',
|
||||
logProjection: {
|
||||
messages: [
|
||||
textLogMessage({
|
||||
uuid: 'runtime-user-evidence',
|
||||
type: 'user',
|
||||
role: 'user',
|
||||
timestamp: '2026-04-21T10:01:00.000Z',
|
||||
sessionId: 'session-from-ledger',
|
||||
content: [{ type: 'text', text: 'Start task-a now' }],
|
||||
}),
|
||||
taskMarkerLogMessage({
|
||||
uuid: 'assistant-start-evidence',
|
||||
parentUuid: 'runtime-user-evidence',
|
||||
timestamp: '2026-04-21T10:02:00.000Z',
|
||||
toolName: 'mcp__agent-teams__task_start',
|
||||
input: { teamName: 'team-a', taskId: 'task-a' },
|
||||
}),
|
||||
taskMarkerLogMessage({
|
||||
uuid: 'assistant-native-evidence',
|
||||
parentUuid: 'assistant-start-evidence',
|
||||
timestamp: '2026-04-21T10:03:00.000Z',
|
||||
toolName: 'bash',
|
||||
input: { command: 'pnpm test' },
|
||||
}),
|
||||
].map((message) => ({
|
||||
...message,
|
||||
sessionId: request.sessionId ?? message.sessionId,
|
||||
})),
|
||||
},
|
||||
})),
|
||||
};
|
||||
const chunkBuilder = {
|
||||
buildBundleChunks: vi.fn((messages) => [
|
||||
{
|
||||
id: 'chunk-ledger-session',
|
||||
kind: 'assistant',
|
||||
messages,
|
||||
},
|
||||
]),
|
||||
};
|
||||
const sessionEvidenceSource = {
|
||||
readTaskRecords: vi.fn(async () => [
|
||||
{
|
||||
taskId: 'task-a',
|
||||
memberName: 'bob',
|
||||
scope: 'member_session_window',
|
||||
laneId: 'lane-from-ledger',
|
||||
sessionId: 'session-from-ledger',
|
||||
since: '2026-04-21T10:00:00.000Z',
|
||||
source: 'delivery_ledger',
|
||||
startMessageUuid: 'runtime-user-evidence',
|
||||
} satisfies OpenCodeTaskLogAttributionRecord,
|
||||
]),
|
||||
};
|
||||
const source = new OpenCodeTaskLogStreamSource(
|
||||
bridge as never,
|
||||
{ resolve: async () => '/tmp/claude' },
|
||||
{
|
||||
getTasks: async () => [createTask({ owner: undefined })],
|
||||
getDeletedTasks: async () => [],
|
||||
} as never,
|
||||
chunkBuilder as never,
|
||||
{ readTaskRecords: vi.fn(async () => []) },
|
||||
sessionEvidenceSource
|
||||
);
|
||||
|
||||
const response = await source.getTaskLogStream('team-a', 'task-a');
|
||||
|
||||
expect(response?.source).toBe('opencode_runtime_attribution');
|
||||
expect(response?.segments[0]?.actor.sessionId).toBe('session-from-ledger');
|
||||
expect(response?.segments[0]?.id).toContain('session-from-ledger');
|
||||
expect(bridge.getOpenCodeTranscript).toHaveBeenCalledWith('/tmp/claude', {
|
||||
teamId: 'team-a',
|
||||
memberName: 'bob',
|
||||
limit: 500,
|
||||
laneId: 'lane-from-ledger',
|
||||
sessionId: 'session-from-ledger',
|
||||
});
|
||||
expect(
|
||||
chunkBuilder.buildBundleChunks.mock.calls[0]?.[0].map(
|
||||
(message: { uuid: string }) => message.uuid
|
||||
)
|
||||
).toEqual(['runtime-user-evidence', 'assistant-start-evidence', 'assistant-native-evidence']);
|
||||
});
|
||||
|
||||
it('sanitizes OpenCode delivery retry envelopes from projected task log text', async () => {
|
||||
const bridge = {
|
||||
getOpenCodeTranscript: vi.fn(async () => ({
|
||||
|
|
|
|||
|
|
@ -0,0 +1,206 @@
|
|||
import { mkdtemp, mkdir, writeFile, rm } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import * as path from 'node:path';
|
||||
|
||||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
|
||||
import { TaskLogOpenCodeSessionEvidenceSource } from '../../../../src/main/services/team/taskLogs/stream/TaskLogOpenCodeSessionEvidenceSource';
|
||||
import {
|
||||
OPENCODE_PROMPT_DELIVERY_LEDGER_SCHEMA_VERSION,
|
||||
type OpenCodePromptDeliveryLedgerRecord,
|
||||
} from '../../../../src/main/services/team/opencode/delivery/OpenCodePromptDeliveryLedger';
|
||||
|
||||
import type { TeamTask } from '../../../../src/shared/types';
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
function createTask(overrides: Partial<TeamTask> = {}): TeamTask {
|
||||
return {
|
||||
id: 'task-a',
|
||||
displayId: 'task-a',
|
||||
subject: 'Implement task',
|
||||
owner: 'bob',
|
||||
status: 'in_progress',
|
||||
createdAt: '2026-04-21T09:00:00.000Z',
|
||||
updatedAt: '2026-04-21T10:00:00.000Z',
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function createLedgerRecord(
|
||||
overrides: Partial<OpenCodePromptDeliveryLedgerRecord> = {}
|
||||
): OpenCodePromptDeliveryLedgerRecord {
|
||||
return {
|
||||
id: 'record-a',
|
||||
teamName: 'team-a',
|
||||
memberName: 'bob',
|
||||
laneId: 'lane-a',
|
||||
runId: 'run-a',
|
||||
runtimeSessionId: 'session-a',
|
||||
inboxMessageId: 'inbox-a',
|
||||
inboxTimestamp: '2026-04-21T10:00:00.000Z',
|
||||
source: 'watcher',
|
||||
messageKind: 'default',
|
||||
replyRecipient: 'user',
|
||||
actionMode: 'do',
|
||||
taskRefs: [
|
||||
{
|
||||
taskId: 'task-a',
|
||||
displayId: 'task-a',
|
||||
teamName: 'team-a',
|
||||
},
|
||||
],
|
||||
payloadHash: 'hash-a',
|
||||
status: 'accepted',
|
||||
responseState: 'pending',
|
||||
attempts: 1,
|
||||
maxAttempts: 3,
|
||||
acceptanceUnknown: false,
|
||||
nextAttemptAt: null,
|
||||
lastAttemptAt: '2026-04-21T10:00:01.000Z',
|
||||
lastObservedAt: null,
|
||||
acceptedAt: '2026-04-21T10:00:02.000Z',
|
||||
respondedAt: null,
|
||||
failedAt: null,
|
||||
inboxReadCommittedAt: null,
|
||||
inboxReadCommitError: null,
|
||||
prePromptCursor: null,
|
||||
postPromptCursor: null,
|
||||
deliveredUserMessageId: 'runtime-user-a',
|
||||
observedAssistantMessageId: null,
|
||||
observedAssistantPreview: null,
|
||||
observedToolCallNames: [],
|
||||
observedVisibleMessageId: null,
|
||||
visibleReplyMessageId: null,
|
||||
visibleReplyInbox: null,
|
||||
visibleReplyCorrelation: null,
|
||||
lastReason: null,
|
||||
diagnostics: [],
|
||||
createdAt: '2026-04-21T10:00:00.000Z',
|
||||
updatedAt: '2026-04-21T10:00:02.000Z',
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
async function writeLedger(input: {
|
||||
teamsBasePath: string;
|
||||
teamName: string;
|
||||
laneId: string;
|
||||
records: OpenCodePromptDeliveryLedgerRecord[];
|
||||
}): Promise<void> {
|
||||
const ledgerPath = path.join(
|
||||
input.teamsBasePath,
|
||||
input.teamName,
|
||||
'.opencode-runtime',
|
||||
'lanes',
|
||||
encodeURIComponent(input.laneId),
|
||||
'opencode-prompt-delivery-ledger.json'
|
||||
);
|
||||
await mkdir(path.dirname(ledgerPath), { recursive: true });
|
||||
await writeFile(
|
||||
ledgerPath,
|
||||
`${JSON.stringify(
|
||||
{
|
||||
schemaVersion: OPENCODE_PROMPT_DELIVERY_LEDGER_SCHEMA_VERSION,
|
||||
updatedAt: '2026-04-21T10:00:00.000Z',
|
||||
data: input.records,
|
||||
},
|
||||
null,
|
||||
2
|
||||
)}\n`
|
||||
);
|
||||
}
|
||||
|
||||
async function createTempTeamsBasePath(): Promise<string> {
|
||||
const dir = await mkdtemp(path.join(tmpdir(), 'task-log-session-evidence-'));
|
||||
tempDirs.push(dir);
|
||||
return dir;
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
describe('TaskLogOpenCodeSessionEvidenceSource', () => {
|
||||
it('returns bounded exact OpenCode session evidence from prompt delivery ledgers', async () => {
|
||||
const teamsBasePath = await createTempTeamsBasePath();
|
||||
await writeLedger({
|
||||
teamsBasePath,
|
||||
teamName: 'team-a',
|
||||
laneId: 'lane-a',
|
||||
records: [
|
||||
createLedgerRecord({
|
||||
id: 'record-old',
|
||||
runtimeSessionId: 'session-old',
|
||||
inboxTimestamp: '2026-04-21T09:00:00.000Z',
|
||||
lastAttemptAt: '2026-04-21T09:00:01.000Z',
|
||||
acceptedAt: '2026-04-21T09:00:01.000Z',
|
||||
createdAt: '2026-04-21T09:00:00.000Z',
|
||||
updatedAt: '2026-04-21T09:00:01.000Z',
|
||||
}),
|
||||
createLedgerRecord({
|
||||
id: 'record-new',
|
||||
runtimeSessionId: 'session-new',
|
||||
deliveredUserMessageId: 'runtime-user-new',
|
||||
inboxTimestamp: '2026-04-21T10:00:00.000Z',
|
||||
lastAttemptAt: '2026-04-21T10:00:01.000Z',
|
||||
acceptedAt: '2026-04-21T10:00:01.000Z',
|
||||
createdAt: '2026-04-21T10:00:00.000Z',
|
||||
updatedAt: '2026-04-21T10:00:01.000Z',
|
||||
}),
|
||||
],
|
||||
});
|
||||
await writeLedger({
|
||||
teamsBasePath,
|
||||
teamName: 'team-a',
|
||||
laneId: 'lane-foreign',
|
||||
records: [
|
||||
createLedgerRecord({
|
||||
id: 'record-foreign-task',
|
||||
laneId: 'lane-foreign',
|
||||
runtimeSessionId: 'session-foreign',
|
||||
taskRefs: [
|
||||
{
|
||||
taskId: 'task-foreign',
|
||||
displayId: 'task-foreign',
|
||||
teamName: 'team-a',
|
||||
},
|
||||
],
|
||||
}),
|
||||
createLedgerRecord({
|
||||
id: 'record-rejected-before-acceptance',
|
||||
laneId: 'lane-foreign',
|
||||
runtimeSessionId: 'session-rejected',
|
||||
status: 'failed_terminal',
|
||||
acceptedAt: null,
|
||||
}),
|
||||
],
|
||||
});
|
||||
|
||||
const source = new TaskLogOpenCodeSessionEvidenceSource({
|
||||
teamsBasePath,
|
||||
maxEvidenceRecords: 1,
|
||||
});
|
||||
|
||||
const records = await source.readTaskRecords('team-a', createTask());
|
||||
|
||||
expect(records).toEqual([
|
||||
expect.objectContaining({
|
||||
taskId: 'task-a',
|
||||
memberName: 'bob',
|
||||
scope: 'member_session_window',
|
||||
laneId: 'lane-a',
|
||||
sessionId: 'session-new',
|
||||
source: 'delivery_ledger',
|
||||
startMessageUuid: 'runtime-user-new',
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('returns an empty candidate list when no matching ledger exists', async () => {
|
||||
const teamsBasePath = await createTempTeamsBasePath();
|
||||
const source = new TaskLogOpenCodeSessionEvidenceSource({ teamsBasePath });
|
||||
|
||||
await expect(source.readTaskRecords('team-a', createTask())).resolves.toEqual([]);
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue