Merge branch 'feat/step6-success-rate' into feat/step3-missing-guidance
This commit is contained in:
@@ -3,7 +3,7 @@ import { ExecutionError, WorkflowPaused } from '../types';
|
||||
import { GraphExecutor } from '../graph-executor';
|
||||
import { graphSchema } from '../lib/schemas';
|
||||
import { createComponentLoader } from '../lib/component-loader';
|
||||
import { writeEvaluation, updateComponentStats } from './execution-evaluator';
|
||||
import { recordComponentStats } from './execution-evaluator';
|
||||
import { parseTriplets } from './triplet-parser';
|
||||
import { searchNodes, type SearchMode, type SearchTarget } from './search-nodes';
|
||||
import { buildExecutionGraph } from './graph-builder';
|
||||
@@ -77,18 +77,8 @@ export async function handleCypherExecute(
|
||||
const result = await executor.execute(parseResult.data as ExecutionGraph, context ?? {}, env.EXEC_CONTEXT);
|
||||
const duration_ms = Date.now() - start;
|
||||
|
||||
// 非同步記錄統計(Phase 7 補充 analytics,目前為 no-op)
|
||||
const componentId = graph.nodes.find(n => n.componentId)?.componentId ?? graphId;
|
||||
const runId = `${graphId}-${Date.now()}`;
|
||||
waitUntil(writeEvaluation(env, {
|
||||
run_id: runId,
|
||||
workflow_id: graphId,
|
||||
component_id: componentId,
|
||||
verdict: 'success',
|
||||
duration_ms,
|
||||
evaluated_at: Date.now(),
|
||||
}));
|
||||
waitUntil(updateComponentStats(env, componentId, 'success', duration_ms));
|
||||
// 非同步回寫每顆零件的執行統計(design.md「執行統計設計」;fire-and-forget 不阻擋回應)
|
||||
waitUntil(recordComponentStats(env, graph.nodes, result.trace));
|
||||
|
||||
return { success: true, data: result.data, trace: result.trace, duration_ms, graph };
|
||||
} catch (err) {
|
||||
@@ -110,19 +100,10 @@ export async function handleCypherExecute(
|
||||
}
|
||||
|
||||
const errMsg = err instanceof Error ? err.message : String(err);
|
||||
const componentId = graph.nodes.find(n => n.componentId)?.componentId ?? graphId;
|
||||
const runId = `${graphId}-${Date.now()}`;
|
||||
waitUntil(writeEvaluation(env, {
|
||||
run_id: runId,
|
||||
workflow_id: graphId,
|
||||
component_id: componentId,
|
||||
verdict: 'failed',
|
||||
duration_ms,
|
||||
error_message: errMsg.slice(0, 200),
|
||||
evaluated_at: Date.now(),
|
||||
}));
|
||||
waitUntil(updateComponentStats(env, componentId, 'failed', duration_ms));
|
||||
// 失敗路徑同樣回寫每顆零件統計:ExecutionError 帶完整 trace(失敗節點有 error、
|
||||
// 之前成功的節點照記成功);非 ExecutionError 無 trace 可歸因 → 不記(誠實:不瞎猜)。
|
||||
if (err instanceof ExecutionError) {
|
||||
waitUntil(recordComponentStats(env, graph.nodes, err.trace));
|
||||
const traceFormatted = err.trace.map(s => ({
|
||||
node: s.nodeId,
|
||||
status: s.error ? 'failed' : 'success',
|
||||
|
||||
@@ -1,36 +1,96 @@
|
||||
/**
|
||||
* Execution Analytics — 零件執行後的統計記錄
|
||||
* Execution Analytics — 零件執行後的統計回寫
|
||||
*
|
||||
* Phase 1 MVP:stub(不寫入任何外部服務)
|
||||
* Phase 7 補充:fire-and-forget POST 至 registry.arcrun.dev/analytics/record
|
||||
* SDD: system-dev/docs/3-specs/arcrun-core-mvp/design.md「執行統計設計」
|
||||
* 執行完成處(cypher-handlers / webhook-handlers 收尾)對本次用到的**每顆零件**
|
||||
* fire-and-forget POST registry `/analytics/record`——統計失敗不影響執行、不增加同步延遲
|
||||
* (呼叫端一律用 waitUntil 包,仿 recordRecipeStats / recordTelemetry 既有慣例)。
|
||||
*
|
||||
* 每顆零件的成敗判定來源=執行 trace(per-node):
|
||||
* - trace step 有 `error` → 失敗(runner throw)
|
||||
* - output 是物件且 `success === false` → 失敗(makeHttpRunner 對非 2xx 不 throw,回這種)
|
||||
* - 其餘 → 成功
|
||||
* FOREACH 重複執行同一節點 → trace 有幾筆就記幾次(每次真實執行都算一次樣本)。
|
||||
*/
|
||||
|
||||
import type { Bindings } from '../types';
|
||||
import type { GraphNode, TraceStep } from '../types';
|
||||
import { wasmWorkerUrl } from '../lib/component-loader';
|
||||
|
||||
export interface EvaluationRecord {
|
||||
run_id: string;
|
||||
workflow_id: string;
|
||||
/** 本模組需要的環境子集(傳整份 Bindings 也相容,仿 SearchNodesEnv 慣例)。 */
|
||||
export type AnalyticsEnv = {
|
||||
WORKER_SUBDOMAIN?: string;
|
||||
/** registry 位置覆蓋(可選;本地 wrangler dev / self-hosted 用)。未設 → wasmWorkerUrl('registry', WORKER_SUBDOMAIN)。 */
|
||||
REGISTRY_BASE_URL?: string;
|
||||
};
|
||||
|
||||
export interface ComponentVerdict {
|
||||
component_id: string;
|
||||
verdict: 'success' | 'failed' | 'timeout';
|
||||
success: boolean;
|
||||
duration_ms: number;
|
||||
error_message?: string;
|
||||
evaluated_at: number;
|
||||
}
|
||||
|
||||
/** 記錄執行結果(MVP:no-op,Phase 7 補充 analytics)*/
|
||||
export async function writeEvaluation(
|
||||
_env: Bindings,
|
||||
_record: EvaluationRecord,
|
||||
): Promise<void> {
|
||||
// Phase 7: POST to registry.arcrun.dev/analytics/record
|
||||
/** 從執行 trace 導出每顆零件的成敗(只算 type=Component 且有 componentId 的節點)。 */
|
||||
export function componentVerdictsFromTrace(
|
||||
nodes: GraphNode[],
|
||||
trace: TraceStep[],
|
||||
): ComponentVerdict[] {
|
||||
const componentByNodeId = new Map<string, string>();
|
||||
for (const n of nodes) {
|
||||
if (n.type === 'Component' && n.componentId) componentByNodeId.set(n.id, n.componentId);
|
||||
}
|
||||
|
||||
const verdicts: ComponentVerdict[] = [];
|
||||
for (const step of trace) {
|
||||
const componentId = componentByNodeId.get(step.nodeId);
|
||||
if (!componentId) continue;
|
||||
|
||||
const out = step.output;
|
||||
const outputSaysFailed =
|
||||
typeof out === 'object' && out !== null && !Array.isArray(out) &&
|
||||
(out as Record<string, unknown>).success === false;
|
||||
|
||||
verdicts.push({
|
||||
component_id: componentId,
|
||||
success: !step.error && !outputSaysFailed,
|
||||
duration_ms: Math.max(0, Number(step.duration_ms) || 0),
|
||||
});
|
||||
}
|
||||
return verdicts;
|
||||
}
|
||||
|
||||
/** 更新零件統計(MVP:no-op,Phase 7 補充)*/
|
||||
export async function updateComponentStats(
|
||||
_env: Bindings,
|
||||
_componentId: string,
|
||||
_verdict: 'success' | 'failed' | 'timeout',
|
||||
_durationMs: number,
|
||||
/**
|
||||
* 對本次執行用到的每顆零件回寫統計到 registry(design.md「Analytics Record」)。
|
||||
* 永不 throw;呼叫端用 waitUntil 包,不阻擋主流程。
|
||||
*/
|
||||
export async function recordComponentStats(
|
||||
env: AnalyticsEnv,
|
||||
nodes: GraphNode[],
|
||||
trace: TraceStep[],
|
||||
): Promise<void> {
|
||||
// Phase 7: update ANALYTICS_KV via registry worker
|
||||
try {
|
||||
const base = (
|
||||
env.REGISTRY_BASE_URL ??
|
||||
(env.WORKER_SUBDOMAIN ? wasmWorkerUrl('registry', env.WORKER_SUBDOMAIN) : undefined)
|
||||
)?.replace(/\/$/, '');
|
||||
if (!base) return;
|
||||
|
||||
const verdicts = componentVerdictsFromTrace(nodes, trace);
|
||||
if (verdicts.length === 0) return;
|
||||
|
||||
await Promise.all(
|
||||
verdicts.map(v =>
|
||||
fetch(`${base}/analytics/record`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
canonical_id: v.component_id,
|
||||
success: v.success,
|
||||
duration_ms: v.duration_ms,
|
||||
}),
|
||||
}).catch(() => undefined), // 統計失敗不影響執行
|
||||
),
|
||||
);
|
||||
} catch {
|
||||
// fire-and-forget:不拋錯,不影響主流程
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,8 @@ import { GraphExecutor } from '../graph-executor';
|
||||
import { graphSchema } from '../lib/schemas';
|
||||
import { createComponentLoader } from '../lib/component-loader';
|
||||
import { recordTelemetry } from '../lib/telemetry';
|
||||
import { recordComponentStats } from './execution-evaluator';
|
||||
import type { GraphNode, TraceStep } from '../types';
|
||||
|
||||
/**
|
||||
* kbdb-base §7.1+§7.5.h:一條工作流執行結束後,把這次用到的 recipe 各記一次成功/失敗到 KBDB 市場星數。
|
||||
@@ -96,6 +98,17 @@ export async function executeWebhookGraph(
|
||||
// kbdb-base §7.1:整體成功 → 用到的 recipe 各記成功一次。
|
||||
recordRecipeStats(env, executor.usedRecipeKeys, true, Date.now(), ctx);
|
||||
|
||||
// arcrun-core-mvp「執行統計設計」:對用到的每顆零件回寫執行結果(fire-and-forget)。
|
||||
{
|
||||
const statsPromise = recordComponentStats(
|
||||
env,
|
||||
(parsed.data as ExecutionGraph).nodes as GraphNode[],
|
||||
result.trace as TraceStep[],
|
||||
);
|
||||
if (ctx?.waitUntil) ctx.waitUntil(statsPromise);
|
||||
else void statsPromise;
|
||||
}
|
||||
|
||||
return { success: true, data: result.data, duration_ms };
|
||||
} catch (err) {
|
||||
const duration_ms = Date.now() - start;
|
||||
@@ -117,6 +130,18 @@ export async function executeWebhookGraph(
|
||||
recordRecipeStats(env, executor.usedRecipeKeys, false, Date.now(), ctx);
|
||||
}
|
||||
|
||||
// 零件統計失敗路徑:ExecutionError 帶完整 trace(失敗節點有 error、先前成功節點照記成功);
|
||||
// paused 非失敗不記;非 ExecutionError 無 trace 可歸因 → 不記。
|
||||
if (!isPaused && err instanceof ExecutionError) {
|
||||
const statsPromise = recordComponentStats(
|
||||
env,
|
||||
(parsed.data as ExecutionGraph).nodes as GraphNode[],
|
||||
err.trace,
|
||||
);
|
||||
if (ctx?.waitUntil) ctx.waitUntil(statsPromise);
|
||||
else void statsPromise;
|
||||
}
|
||||
|
||||
if (err instanceof ExecutionError) {
|
||||
const traceFormatted = err.trace.map(s => ({
|
||||
node: s.nodeId,
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
// 單元測試:execution-evaluator — 從 trace 導出每顆零件成敗 + 回寫 registry
|
||||
// SDD: system-dev/docs/3-specs/arcrun-core-mvp/design.md「執行統計設計」
|
||||
|
||||
import { describe, it, expect, vi, afterEach } from 'vitest';
|
||||
import { componentVerdictsFromTrace, recordComponentStats } from '../src/actions/execution-evaluator';
|
||||
import type { GraphNode, TraceStep } from '../src/types';
|
||||
|
||||
const NODES: GraphNode[] = [
|
||||
{ id: 'input', type: 'Input' },
|
||||
{ id: 'fetch', type: 'Component', componentId: 'http_request' },
|
||||
{ id: 'transform', type: 'Component', componentId: 'code' },
|
||||
{ id: 'output', type: 'Output' },
|
||||
];
|
||||
|
||||
function step(nodeId: string, over: Partial<TraceStep> = {}): TraceStep {
|
||||
return { nodeId, type: 'Component', input: {}, output: { ok: true }, duration_ms: 10, ...over };
|
||||
}
|
||||
|
||||
describe('componentVerdictsFromTrace', () => {
|
||||
it('只算 Component 節點;Input/Output 跳過', () => {
|
||||
const verdicts = componentVerdictsFromTrace(NODES, [
|
||||
step('input', { type: 'Input' }),
|
||||
step('fetch'),
|
||||
step('output', { type: 'Output' }),
|
||||
]);
|
||||
expect(verdicts).toEqual([{ component_id: 'http_request', success: true, duration_ms: 10 }]);
|
||||
});
|
||||
|
||||
it('trace 有 error → 該零件記失敗', () => {
|
||||
const verdicts = componentVerdictsFromTrace(NODES, [
|
||||
step('fetch', { error: 'boom', output: null }),
|
||||
]);
|
||||
expect(verdicts).toEqual([{ component_id: 'http_request', success: false, duration_ms: 10 }]);
|
||||
});
|
||||
|
||||
it('output.success === false → 記失敗(makeHttpRunner 對非 2xx 不 throw)', () => {
|
||||
const verdicts = componentVerdictsFromTrace(NODES, [
|
||||
step('fetch', { output: { success: false, status: 500, error: 'oops' } }),
|
||||
]);
|
||||
expect(verdicts[0].success).toBe(false);
|
||||
});
|
||||
|
||||
it('FOREACH 同節點多筆 trace → 每次執行各記一次樣本', () => {
|
||||
const verdicts = componentVerdictsFromTrace(NODES, [
|
||||
step('fetch'),
|
||||
step('fetch', { error: 'x', output: null }),
|
||||
step('fetch'),
|
||||
]);
|
||||
expect(verdicts).toHaveLength(3);
|
||||
expect(verdicts.map(v => v.success)).toEqual([true, false, true]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('recordComponentStats', () => {
|
||||
afterEach(() => vi.unstubAllGlobals());
|
||||
|
||||
it('對每顆零件各發一次 POST /analytics/record(fire-and-forget)', async () => {
|
||||
const calls: Array<{ url: string; body: Record<string, unknown> }> = [];
|
||||
vi.stubGlobal('fetch', vi.fn(async (url: string, init: RequestInit) => {
|
||||
calls.push({ url: String(url), body: JSON.parse(String(init.body)) });
|
||||
return new Response('{}', { status: 200 });
|
||||
}));
|
||||
|
||||
await recordComponentStats(
|
||||
{ REGISTRY_BASE_URL: 'http://registry.local' },
|
||||
NODES,
|
||||
[step('fetch'), step('transform', { error: 'bad', output: null })],
|
||||
);
|
||||
|
||||
expect(calls).toHaveLength(2);
|
||||
expect(calls[0].url).toBe('http://registry.local/analytics/record');
|
||||
expect(calls[0].body).toEqual({ canonical_id: 'http_request', success: true, duration_ms: 10 });
|
||||
expect(calls[1].body).toEqual({ canonical_id: 'code', success: false, duration_ms: 10 });
|
||||
});
|
||||
|
||||
it('registry 打不到也不 throw(統計失敗不影響執行)', async () => {
|
||||
vi.stubGlobal('fetch', vi.fn(async () => { throw new Error('network down'); }));
|
||||
await expect(
|
||||
recordComponentStats({ REGISTRY_BASE_URL: 'http://registry.local' }, NODES, [step('fetch')]),
|
||||
).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it('無 REGISTRY_BASE_URL 也無 WORKER_SUBDOMAIN → 靜默略過不打', async () => {
|
||||
const fetchSpy = vi.fn();
|
||||
vi.stubGlobal('fetch', fetchSpy);
|
||||
await recordComponentStats({}, NODES, [step('fetch')]);
|
||||
expect(fetchSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('未設 REGISTRY_BASE_URL → 用 wasmWorkerUrl 慣例組 registry URL', async () => {
|
||||
const calls: string[] = [];
|
||||
vi.stubGlobal('fetch', vi.fn(async (url: string) => {
|
||||
calls.push(String(url));
|
||||
return new Response('{}', { status: 200 });
|
||||
}));
|
||||
await recordComponentStats({ WORKER_SUBDOMAIN: 'uncle6-me' }, NODES, [step('fetch')]);
|
||||
expect(calls[0]).toBe('https://arcrun-registry.uncle6-me.workers.dev/analytics/record');
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user