diff --git a/dashboard/package.json b/dashboard/package.json index 92861bea..f3888ed1 100644 --- a/dashboard/package.json +++ b/dashboard/package.json @@ -1,6 +1,6 @@ { "name": "workstacean-dashboard", - "version": "0.7.12", + "version": "0.7.13", "private": true, "type": "module", "scripts": { diff --git a/lib/plugins/github.ts b/lib/plugins/github.ts index 01345120..6901d967 100644 --- a/lib/plugins/github.ts +++ b/lib/plugins/github.ts @@ -328,6 +328,7 @@ export class GitHubPlugin implements Plugin { const content = String((msg.payload as Record).content ?? "").trim(); if (!content) return; + if (/^Skill ".+" completed by \w+$/i.test(content)) return; await this._postComment(getToken, pending, content); }); @@ -552,6 +553,10 @@ export class GitHubPlugin implements Plugin { console.log(`[github] ${event} on ${ctx.owner}/${ctx.repo}#${ctx.number} → ${skillHint ?? "default"}`); } + /** Tracks recently-dispatched (owner/repo#number) to suppress duplicate webhook deliveries. */ + private recentDispatches = new Map(); + private static readonly DEDUP_WINDOW_MS = 60_000; + private _handleAutoTriage( event: string, payload: Record, @@ -560,6 +565,14 @@ export class GitHubPlugin implements Plugin { bus: EventBus, getToken: (owner: string, repo: string) => Promise, ): void { + const dedupKey = `${ctx.owner}/${ctx.repo}#${ctx.number}`; + const lastDispatched = this.recentDispatches.get(dedupKey); + if (lastDispatched && Date.now() - lastDispatched < GitHubPlugin.DEDUP_WINDOW_MS) { + console.log(`[github] Auto-triage: skipping duplicate ${dedupKey} (dispatched ${Date.now() - lastDispatched}ms ago)`); + return; + } + this.recentDispatches.set(dedupKey, Date.now()); + (async () => { try { const token = await getToken(ctx.owner, ctx.repo); diff --git a/package.json b/package.json index ffda457a..b00929e9 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "workstacean", - "version": "0.7.12", + "version": "0.7.13", "type": "module", "scripts": { "start": "bun run src/index.ts", diff --git a/src/executor/executors/a2a-executor.ts b/src/executor/executors/a2a-executor.ts index adde0f6d..7cb20953 100644 --- a/src/executor/executors/a2a-executor.ts +++ b/src/executor/executors/a2a-executor.ts @@ -152,6 +152,11 @@ export class A2AExecutor implements IExecutor { return this.config.name; } + /** Expose the configured agent URL so the dispatcher can pick an appropriate callback base URL. */ + get url(): string { + return this.config.url; + } + /** * Update the streaming + pushNotifications capability flags from the agent * card. Called by SkillBrokerPlugin after every card discovery pass so the @@ -458,6 +463,16 @@ export class A2AExecutor implements IExecutor { const delta = this._worldStateDeltaFromParts(artifact.parts); if (delta) streamDeltaData[WORLDSTATE_DELTA_MIME_TYPE] = delta; } + // Non-terminal state (submitted / working / input-required) on the + // initial task event → stop consuming the SSE stream and let + // TaskTracker drive the task to completion via tasks/get polling. + // Holding the stream open longer forces slow agents to keep their + // SSE generator alive for minutes, which crashes them when the + // sync caller eventually times out and closes the connection. + if (taskState !== "completed" && taskState !== "failed" + && taskState !== "canceled" && taskState !== "rejected") { + break; + } continue; } if (event.kind === "status-update") { diff --git a/src/executor/extensions/blast.ts b/src/executor/extensions/blast.ts index b7fe2010..1251a282 100644 --- a/src/executor/extensions/blast.ts +++ b/src/executor/extensions/blast.ts @@ -102,6 +102,13 @@ export class BlastRegistry { this.byKey.clear(); } + clearAgent(agentName: string): void { + const prefix = `${agentName}::`; + for (const key of this.byKey.keys()) { + if (key.startsWith(prefix)) this.byKey.delete(key); + } + } + get size(): number { return this.byKey.size; } diff --git a/src/executor/skill-dispatcher-plugin.ts b/src/executor/skill-dispatcher-plugin.ts index 3bf194ca..5c2895d1 100644 --- a/src/executor/skill-dispatcher-plugin.ts +++ b/src/executor/skill-dispatcher-plugin.ts @@ -361,7 +361,15 @@ export class SkillDispatcherPlugin implements Plugin { // burn a round-trip on every long-running task against every agent, // most of which reject. SkillBrokerPlugin refreshes the flag every // 10 min so capability changes land automatically. - const callbackBaseUrl = process.env.WORKSTACEAN_BASE_URL; + // + // Callback URL routing: + // - Docker-internal agents (hostname like `http://quinn:7870`) use + // WORKSTACEAN_INTERNAL_BASE_URL (default http://workstacean:3000) + // which resolves inside the shared docker network. + // - External agents reached over Tailscale / public networks + // (hostname has a dot, e.g. `http://host.tailnet.ts.net:...`) + // use WORKSTACEAN_BASE_URL — the operator-configured public URL. + const callbackBaseUrl = this._pickCallbackBaseUrl(a2aExecutor.url); if (callbackBaseUrl && a2aExecutor.pushNotifications) { const callbackUrl = `${callbackBaseUrl.replace(/\/$/, "")}/api/a2a/callback/${encodeURIComponent(taskId)}`; void a2aExecutor.registerPushNotification(taskId, callbackUrl, callbackToken, correlationId, parentId) @@ -601,6 +609,29 @@ export class SkillDispatcherPlugin implements Plugin { }); } + /** + * Pick the callback base URL for push notifications based on whether the + * target agent is docker-internal or external. Docker service names are + * short hostnames (no dots); external hosts use FQDNs or IPs. + * + * Internal default: http://workstacean:3000 (resolves on shared docker net). + * External default: process.env.WORKSTACEAN_BASE_URL. + */ + private _pickCallbackBaseUrl(agentUrl: string | undefined): string | undefined { + if (!agentUrl) return process.env.WORKSTACEAN_BASE_URL; + try { + const { hostname } = new URL(agentUrl); + // Docker service names are single-label (no dot, not an IP). + const isDockerInternal = !hostname.includes(".") && !hostname.includes(":"); + if (isDockerInternal) { + return process.env.WORKSTACEAN_INTERNAL_BASE_URL ?? "http://workstacean:3000"; + } + return process.env.WORKSTACEAN_BASE_URL; + } catch { + return process.env.WORKSTACEAN_BASE_URL; + } + } + private async _fileTriageOnBoard( github: { title: string; owner?: string; repo?: string; number?: number; url?: string }, triageSummary: string | undefined, diff --git a/src/plugins/skill-broker-plugin.ts b/src/plugins/skill-broker-plugin.ts index 78dacf20..ce5f46f9 100644 --- a/src/plugins/skill-broker-plugin.ts +++ b/src/plugins/skill-broker-plugin.ts @@ -31,6 +31,11 @@ import { HITL_MODE_URI, type HitlMode, } from "../executor/extensions/hitl-mode.ts"; +import { + defaultBlastRegistry, + BLAST_URI, + type BlastRadius, +} from "../executor/extensions/blast.ts"; interface AgentSkill { name: string; @@ -269,6 +274,7 @@ export class SkillBrokerPlugin implements Plugin { } this._loadHitlModeDeclarations(agent.name, card); + this._loadBlastDeclarations(agent.name, card); } catch (err) { console.debug(`[skill-broker] ${agent.name}: card discovery skipped:`, err instanceof Error ? err.message : err); } @@ -312,6 +318,36 @@ export class SkillBrokerPlugin implements Plugin { } } + private _loadBlastDeclarations(agentName: string, card: AgentCard): void { + defaultBlastRegistry.clearAgent(agentName); + + const ext = (card.capabilities?.extensions ?? []).find(e => e?.uri === BLAST_URI); + if (!ext) return; + + const params = (ext.params ?? {}) as { skills?: Record }; + const entries = params.skills && typeof params.skills === "object" ? params.skills : {}; + const validRadii = new Set(["self", "project", "repo", "fleet", "public"]); + let declared = 0; + + for (const [skill, rawEntry] of Object.entries(entries)) { + const entry = rawEntry as { radius?: unknown; note?: unknown } | undefined; + if (!entry || typeof entry !== "object") continue; + if (typeof entry.radius !== "string" || !validRadii.has(entry.radius)) continue; + + defaultBlastRegistry.declare({ + agentName, + skill, + radius: entry.radius as BlastRadius, + ...(typeof entry.note === "string" ? { note: entry.note } : {}), + }); + declared++; + } + + if (declared > 0) { + console.log(`[skill-broker] ${agentName}: loaded ${declared} blast declaration(s)`); + } + } + private async _fetchCard(url: string): Promise { const baseUrl = url.replace(/\/a2a\/?$/, ""); const factory = new ClientFactory({