wujianhui/sw-ads-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(swads): make daily report generation resilient

Browse Source
This commit is contained in:
Jeffrey Wu
2026-09-09 17:59:05 +08:00
parent f4e3f41c2c
commit 8361b10c66
13 changed files with 278 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/harness/src/session.ts
+71 -5
View File
@@ -11,7 +11,7 @@ import {
type AgentSessionEvent,
type InlineExtension,
} from "@earendil-works/pi-coding-agent";
import type { ModelConnectionInput, SendMessageInput } from "@agent-studio/shared";
import type { ModelConnectionInput, ReportArtifact, SendMessageInput } from "@agent-studio/shared";
import { InMemoryApprovalBroker, type ApprovalBroker } from "./approval.js";
import type { FileAttachmentStore } from "./attachments.js";
import { HarnessError, redactValue, redactServerSecret, safeErrorMessage } from "./errors.js";
@@ -43,13 +43,15 @@ export interface AgentSessionFactoryInput {
skillRequested: (skill: SkillIndexEntry, toolCallId: string) => void;
skillLoaded: (skill: SkillIndexEntry, toolCallId: string) => void;
toolRequested: (toolCallId: string, toolName: string, argumentsValue: unknown) => void;
reportGenerated: (toolCallId: string, artifact: ReportArtifact, markdown: string) => void;
toolFailed: (toolCallId: string, toolName: string, error: string) => void;
}
export interface AgentSessionFactory { create(input: AgentSessionFactoryInput): Promise<AgentSessionPort> }
class PiAgentSessionPort implements AgentSessionPort {
readonly modelId: string;
readonly supportsImages: boolean;
constructor(private readonly session: AgentSession) {
constructor(private readonly session: AgentSession, private readonly completeRun?: () => Promise<string | undefined> | string | undefined) {
if (!session.model) throw new Error("Pi AgentSession was created without an explicit model");
this.modelId = session.model.id;
this.supportsImages = session.model.input.includes("image");
@@ -57,6 +59,11 @@ class PiAgentSessionPort implements AgentSessionPort {
async prompt(text: string, options?: { images?: unknown[] }): Promise<void> {
const images = options?.images as ImageContent[] | undefined;
await this.session.prompt(text, images ? { images } : undefined);
// Pi reports provider failures through its message stream and resolves prompt().
// Convert a terminal model error into Harness run.failed instead of run.completed.
if (this.session.agent.state.errorMessage) throw new Error("Model provider run failed");
const completionError = await this.completeRun?.();
if (completionError) throw new Error(completionError);
}
abort(): Promise<void> { return this.session.abort(); }
subscribe(listener: (event: unknown) => void): () => void { return this.session.subscribe(listener); }
@@ -76,8 +83,23 @@ export class PiAgentSessionFactory implements AgentSessionFactory {
async create(input: AgentSessionFactoryInput): Promise<AgentSessionPort> {
const resolved = await this.models.createRuntime(input.sessionId);
const swads = new SwadsReportTools(this.gateway, this.reports, input.getRun);
let swadsWorkflow: { runId: string; rendered: boolean; queryCount: number; finalTextSeen: boolean; fatal: boolean; explained: boolean } | undefined;
const extension: InlineExtension = { name: "agent-studio-safety", hidden: true, factory: (pi) => {
const pendingSkills = new Map<string, SkillIndexEntry>();
pi.on("before_agent_start", (event) => {
const run = input.getRun();
swadsWorkflow = run && event.prompt.includes('<skill name="swads-daily-report"') ? { runId: run.runId, rendered: false, queryCount: 0, finalTextSeen: false, fatal: false, explained: false } : undefined;
});
pi.on("message_end", (event) => {
const run = input.getRun();
if (!run || swadsWorkflow?.runId !== run.runId || event.message.role !== "assistant") return;
if (swadsWorkflow.queryCount > 0 && assistantText(event.message).trim()) swadsWorkflow.finalTextSeen = true;
});
pi.on("turn_end", (event) => {
const run = input.getRun(); const workflow = swadsWorkflow;
if (!run || workflow?.runId !== run.runId || workflow.rendered || event.message.role !== "assistant") return;
if (workflow.fatal && assistantText(event.message).trim()) workflow.explained = true;
});
pi.on("tool_call", async (event) => {
const run = input.getRun();
if (!run) return { block: true, reason: "No active Harness run" };
@@ -99,6 +121,7 @@ export class PiAgentSessionFactory implements AgentSessionFactory {
if (typeof maybePath === "string") await input.workspace.assertReadPath(maybePath);
}
input.toolRequested(event.toolCallId, event.toolName, event.toolName === "render_swads_daily_report" ? { status: "requested" } : publicToolArguments(event.input, input.cwd, matchedSkill));
if (event.toolName === "render_swads_daily_report" && swadsWorkflow?.runId === run.runId && !swadsWorkflow.finalTextSeen) return { block: true, reason: "Final Markdown analysis is required before rendering the SW Ads report" };
if (event.toolName === "save_report_draft") {
const decision = await this.approvals.request({ sessionId: input.sessionId, runId: run.runId, toolCallId: event.toolCallId, toolName: event.toolName, arguments: event.input, riskLevel: "medium" }, run.signal);
if (decision.status !== "approved") return { block: true, reason: `Human approval ${decision.status}` };
@@ -107,6 +130,15 @@ export class PiAgentSessionFactory implements AgentSessionFactory {
});
pi.on("tool_result", async (event) => {
const run = input.getRun();
if (run && swadsWorkflow?.runId === run.runId) {
const toolInput = objectValue(event.input);
if (event.toolName === "swads_cli" && !event.isError && toolInput.command === "metrics.query") swadsWorkflow.queryCount++;
if (event.toolName === "swads_cli" && event.isError && ["capabilities", "whoami", "metrics.catalog", "metrics.query"].includes(String(toolInput.command))) swadsWorkflow.fatal = true;
if (event.toolName === "render_swads_daily_report") {
if (event.isError) swadsWorkflow.fatal = true;
else swadsWorkflow.rendered = true;
}
}
const skill = pendingSkills.get(event.toolCallId);
if (run && skill && !event.isError && event.toolName === "read") {
input.skillLoaded(skill, event.toolCallId);
@@ -126,7 +158,23 @@ export class PiAgentSessionFactory implements AgentSessionFactory {
sessionManager: SessionManager.inMemory(input.cwd),
settingsManager: SettingsManager.inMemory(),
});
return new PiAgentSessionPort(session);
return new PiAgentSessionPort(session, async () => {
const run = input.getRun(); const workflow = swadsWorkflow;
if (!run || workflow?.runId !== run.runId || workflow.rendered || workflow.explained) return undefined;
if (!workflow.fatal && workflow.queryCount >= 4) {
const toolCallId = `swads-fallback-${randomUUID()}`;
input.toolRequested(toolCallId, "render_swads_daily_report", { status: "server-fallback" });
try {
const result = await swads.renderCollected(run.signal);
workflow.rendered = true;
input.reportGenerated(toolCallId, result.artifact, result.markdown);
return undefined;
} catch (error) {
input.toolFailed(toolCallId, "render_swads_daily_report", error instanceof HarnessError ? error.code : "SWADS_REPORT_FAILED");
}
}
return "SWADS_REPORT_INCOMPLETE";
});
}
}
@@ -157,7 +205,7 @@ export class InMemorySessionRegistry implements SessionRegistry {
private readonly catalog: SkillCatalog,
private factory: AgentSessionFactory | undefined,
approvalTimeoutMs = 60_000,
private readonly runTimeoutMs = 120_000,
private readonly runTimeoutMs = 300_000,
) {
this.approvals = new InMemoryApprovalBroker(approvalTimeoutMs, (resolution) => {
this.events.append(resolution.request.sessionId, { type: "approval.resolved", runId: resolution.request.runId, payload: { approvalId: resolution.request.approvalId, toolCallId: resolution.request.toolCallId, status: resolution.decision.status, resolvedAt: resolution.decision.resolvedAt } });
@@ -220,7 +268,7 @@ export class InMemorySessionRegistry implements SessionRegistry {
this.events.append(id, { type: "user.message", runId, payload: { messageId: `msg_${randomUUID()}`, text: message.text, attachmentIds: message.attachmentIds } });
this.events.append(id, { type: "run.started", runId, payload: { modelId: agent.modelId } });
const images: ImageContent[] = await Promise.all(selected.map(async (item) => ({ type: "image" as const, mimeType: item.mediaType, data: await this.attachments.readBase64(id, item.attachmentId) })));
session.timeout = setTimeout(() => void this.cancel(id, "timeout"), this.runTimeoutMs);
this.touchRunTimeout(session);
if (expanded.skill) {
for (const type of ["skill.requested", "skill.loaded"] as const) this.events.append(id, { type, runId, payload: { name: expanded.skill.name, filePath: expanded.skill.filePath, source: "command" } });
}
@@ -239,12 +287,15 @@ export class InMemorySessionRegistry implements SessionRegistry {
skillRequested: (skill, toolCallId) => { const run = session.currentRun; if (run) this.events.append(session.sessionId, { type: "skill.requested", runId: run.runId, payload: { name: skill.name, filePath: skill.filePath, toolCallId, source: "read" } }); },
skillLoaded: (skill, toolCallId) => { const run = session.currentRun; if (run) this.events.append(session.sessionId, { type: "skill.loaded", runId: run.runId, payload: { name: skill.name, filePath: skill.filePath, toolCallId, source: "read" } }); },
toolRequested: (toolCallId, toolName, argumentsValue) => { const run = session.currentRun; if (run && !session.requestedToolCalls.has(toolCallId)) { session.requestedToolCalls.add(toolCallId); this.events.append(session.sessionId, { type: "tool.requested", runId: run.runId, payload: { toolCallId, toolName, arguments: redactValue(argumentsValue) } }); } },
reportGenerated: (toolCallId, artifact, markdown) => { const run = session.currentRun; if (run) { this.events.append(session.sessionId, { type: "tool.completed", runId: run.runId, payload: { toolCallId, toolName: "render_swads_daily_report", result: artifact } }); this.events.append(session.sessionId, { type: "report.generated", runId: run.runId, payload: { assistantMessageId: run.assistantMessageId, artifact, markdown } }); } },
toolFailed: (toolCallId, toolName, error) => { const run = session.currentRun; if (run) this.events.append(session.sessionId, { type: "tool.failed", runId: run.runId, payload: { toolCallId, toolName, error } }); },
});
session.unsubscribe = session.agent.subscribe((event) => this.onPiEvent(session, event as AgentSessionEvent));
}
private onPiEvent(session: InternalSession, event: AgentSessionEvent): void {
const run = session.currentRun;
if (!run) return;
this.touchRunTimeout(session);
// The blocking Extension preflight is the single source for tool.requested.
// Pi's later execution-start event has the same toolCallId and would create
// duplicate assistant-ui Tool parts (and can expose pre-normalized paths).
@@ -266,6 +317,15 @@ export class InMemorySessionRegistry implements SessionRegistry {
if (session.currentRun?.runId === run.runId) session.currentRun = undefined;
}
}
private touchRunTimeout(session: InternalSession): void {
const run = session.currentRun;
if (!run || !session.busy || run.controller.signal.aborted) return;
if (session.timeout) clearTimeout(session.timeout);
const runId = run.runId;
session.timeout = setTimeout(() => {
if (session.currentRun?.runId === runId) void this.cancel(session.sessionId, "timeout");
}, this.runTimeoutMs);
}
async cancel(id: string, reason: "user" | "timeout" | "session_deleted" | "shutdown" = "user"): Promise<void> {
const session = this.internal(id);
const run = session.currentRun;
@@ -296,6 +356,12 @@ export class InMemorySessionRegistry implements SessionRegistry {
private async disposeAgent(session: InternalSession): Promise<void> { session.unsubscribe?.(); session.unsubscribe = undefined; session.agent?.dispose(); session.agent = undefined; }
}
function objectValue(value: unknown): Record<string, unknown> { return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {}; }
function assistantText(message: unknown): string {
const content = objectValue(message).content;
return Array.isArray(content) ? content.filter((part) => objectValue(part).type === "text").map((part) => String(objectValue(part).text ?? "")).join("\n") : "";
}
function publicToolArguments(value: unknown, cwd: string, skill?: SkillIndexEntry): unknown {
const redacted = redactValue(value);
if (!redacted || typeof redacted !== "object" || Array.isArray(redacted)) return redacted;