Add A2A worker progress updates

.github/workflows/live-a2a.yml

@@ -1,6 +1,6 @@
 name: Live — Agent2Agent
 
-# Four real protocol legs cover both roles and conversation lengths:
+# Real protocol legs cover both roles, conversation lengths, and worker progress:
 # inbound/outbound × single-turn/multi-turn. The plugin and remote identities
 # are preconfigured to allow one another in both directions.
 on:
@@ -42,6 +42,7 @@ jobs:
         scenario:
           - inbound-single
           - inbound-multi
+          - inbound-progress
           - outbound-single
           - outbound-multi
 
@@ -72,15 +73,24 @@ jobs:
           bash "$GITHUB_WORKSPACE/tests/ci/npm_with_retry.sh" install -g openclaw@latest
           openclaw --version
 
-      - name: Configure the real model
+      - name: Configure model
         env:
           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
         run: |
+          if [ "${{ matrix.scenario }}" = "inbound-progress" ]; then
+            BASE_URL="http://127.0.0.1:8088/v1"
+            MODEL_ID="mock-model"
+            API_KEY="sk-mock-not-used"
+          else
+            BASE_URL="https://api.openai.com/v1"
+            MODEL_ID="gpt-5.6-sol"
+            API_KEY="$OPENAI_API_KEY"
+          fi
           openclaw onboard --non-interactive \
             --auth-choice custom-api-key \
-            --custom-base-url "https://api.openai.com/v1" \
-            --custom-model-id "gpt-5.6-sol" \
-            --custom-api-key "$OPENAI_API_KEY" \
+            --custom-base-url "$BASE_URL" \
+            --custom-model-id "$MODEL_ID" \
+            --custom-api-key "$API_KEY" \
             --custom-compatibility openai \
             --secret-input-mode plaintext \
             --skip-health \
@@ -104,10 +114,25 @@ jobs:
           openclaw config set channels.inkbox.apiKey "$OPENCLAW_INKBOX_API_KEY"
           openclaw config set channels.inkbox.identity "$HANDLE"
           openclaw config set channels.inkbox.voicemailDetection disabled
+          openclaw config set channels.inkbox.a2aProgressIntervalSeconds 60 --strict-json
           openclaw config set channels.inkbox.signingKey "$OPENCLAW_INKBOX_SIGNING_KEY"
           openclaw config set tools.allow '["inkbox"]' --strict-json
           openclaw config set tools.profile full
 
+      - name: Start deterministic progress model
+        if: matrix.scenario == 'inbound-progress'
+        env:
+          MOCK_A2A_SCENARIO: inbound-progress
+        run: |
+          nohup python3 "$GITHUB_WORKSPACE/tests/live/mock_openai.py" 8088 > "$RUNNER_TEMP/mock.log" 2>&1 &
+          echo $! > "$RUNNER_TEMP/mock.pid"
+          for _ in $(seq 1 10); do
+            curl -sf --connect-timeout 1 --max-time 3 http://127.0.0.1:8088/v1/models >/dev/null && exit 0
+            sleep 1
+          done
+          echo "::error::deterministic progress model did not start"
+          exit 1
+
       - name: Start gateway
         env:
           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
@@ -153,3 +178,4 @@ jobs:
         if: always()
         run: |
           kill "$(cat "$RUNNER_TEMP/gateway.pid" 2>/dev/null)" 2>/dev/null || true
+          kill "$(cat "$RUNNER_TEMP/mock.pid" 2>/dev/null)" 2>/dev/null || true

README.md

@@ -394,6 +394,7 @@ After the gateway prints `[gateway] ready`, `[inkbox] tunnel open`, mail/text su
 | `allowedRecipients` | no | - | Outbound recipient allowlist for messaging targets and A2A Agent Card URLs. Empty means no local outbound filtering. |
 | `allowedInboundContactIds` | no | - | Optional local inbound allowlist by Inkbox contact UUID. Empty means Inkbox contact rules decide reachability. |
 | `includeContactMemories` | no | `true` | Include memories from the matched contact as background context for inbound email, messaging, reactions, and calls. Set `false` to disable them. |
+| `a2aProgressIntervalSeconds` | no | `180` | Send a short nonterminal progress update while serving an A2A task at this cadence. Set to `0` to disable periodic updates. The immediate receipt includes the configured frequency. |
 | `sms.batchDelayMs` | no | `0` | Inbound SMS and iMessage fragment batching window. |
 | `voiceStack` | no | legacy-compatible | `inkbox_voice_ai`, `openai_realtime`, or `inkbox_tts_stt`. Setup always writes an explicit value. |
 | `voiceAiAuthorityMode` | Voice AI | saved server value | Informational local copy of the selected `contact_scoped` or `yolo` authority. |

index.ts

@@ -33,6 +33,10 @@ import {
   recordHostedSmsAfterToolCall,
   recordHostedSmsBeforeToolCall,
 } from "./src/hosted-call-tool-settlement.js";
+import {
+  bindA2AProgressActivityToRun,
+  recordA2AProgressToolActivity,
+} from "./src/a2a-progress-activity.js";
 
 type OpenClawChannelEntry = {
   id: string;
@@ -156,8 +160,15 @@ function registerInkboxTools(api: any): void {
 }
 
 function registerHostedCallSettlementHooks(api: any): void {
-  api.on("before_agent_run", bindHostedSmsCaptureToRun);
-  api.on("before_tool_call", recordHostedSmsBeforeToolCall);
+  api.on("before_agent_run", (event: any, context: any) => {
+    bindHostedSmsCaptureToRun(event, context);
+    bindA2AProgressActivityToRun(event, context);
+  });
+  api.on("before_tool_call", async (event: any, context: any) => {
+    const decision = await recordHostedSmsBeforeToolCall(event, context);
+    if (!decision?.block) recordA2AProgressToolActivity(event, context);
+    return decision;
+  });
   api.on("after_tool_call", recordHostedSmsAfterToolCall);
   api.on("model_call_ended", recordHostedModelCallEnded);
 }

openclaw.plugin.json

@@ -143,6 +143,11 @@
             "minimum": 1,
             "description": "Maximum time to let the hidden voice agent warmup run before aborting. Defaults to 70000."
           },
+          "a2aProgressIntervalSeconds": {
+            "type": "integer",
+            "minimum": 0,
+            "description": "Seconds between short progress updates while serving an A2A task. Set to 0 to disable periodic updates. Defaults to 180."
+          },
           "allowedRecipients": {
             "type": "array",
             "items": { "type": "string" }
@@ -236,6 +241,7 @@
                 "voiceAgentPrewarm": { "type": "boolean" },
                 "voiceAgentPrewarmTtlMs": { "type": "integer", "minimum": 0 },
                 "voiceAgentPrewarmTimeoutMs": { "type": "integer", "minimum": 1 },
+                "a2aProgressIntervalSeconds": { "type": "integer", "minimum": 0 },
                 "allowedRecipients": {
                   "type": "array",
                   "items": { "type": "string" }
@@ -417,6 +423,11 @@
         "minimum": 1,
         "description": "Maximum time to let the hidden voice agent warmup run before aborting. Defaults to 70000."
       },
+      "a2aProgressIntervalSeconds": {
+        "type": "integer",
+        "minimum": 0,
+        "description": "Seconds between short progress updates while serving an A2A task. Set to 0 to disable periodic updates. Defaults to 180."
+      },
       "vault": {
         "type": "object",
         "additionalProperties": false,

src/a2a-context.ts

@@ -3,6 +3,7 @@ export interface ActiveA2ATurn {
   messageId: string;
   contextId: string;
   replyIntentCommitted: boolean;
+  beforeReplyIntent?: () => Promise<void>;
 }
 
 const active = new Map<string, ActiveA2ATurn>();

src/a2a-progress-activity.ts

@@ -0,0 +1,75 @@
+interface HookContext {
+  sessionKey?: string;
+  runId?: string;
+}
+
+interface ToolHookEvent {
+  toolName?: string;
+  runId?: string;
+}
+
+interface ActivityCapture {
+  promptMarker: string;
+  runId?: string;
+  toolIdentifiers: string[];
+}
+
+const captures = new Map<string, ActivityCapture>();
+const MAX_TOOL_IDENTIFIERS = 8;
+const MAX_TOOL_IDENTIFIER_CHARS = 80;
+
+export function normalizeA2AIdentifierText(value: unknown): string {
+  return String(value ?? "")
+    .trim()
+    .toLowerCase()
+    .replace(/[^a-z0-9_.:-]+/g, "_")
+    .replace(/^[_.:-]+|[_.:-]+$/g, "");
+}
+
+export function normalizeA2AToolIdentifier(value: unknown): string {
+  return normalizeA2AIdentifierText(value)
+    .slice(0, MAX_TOOL_IDENTIFIER_CHARS)
+    .replace(/[_.:-]+$/g, "");
+}
+
+export function beginA2AProgressActivityCapture(params: {
+  sessionKey: string;
+  promptMarker: string;
+}): { snapshot(): string[]; finish(): void } {
+  const capture: ActivityCapture = {
+    promptMarker: params.promptMarker,
+    toolIdentifiers: [],
+  };
+  captures.set(params.sessionKey, capture);
+  return {
+    snapshot: () => [...capture.toolIdentifiers],
+    finish: () => {
+      if (captures.get(params.sessionKey) === capture) captures.delete(params.sessionKey);
+    },
+  };
+}
+
+export function bindA2AProgressActivityToRun(
+  event: { prompt?: string },
+  context: HookContext,
+): void {
+  const capture = context.sessionKey ? captures.get(context.sessionKey) : undefined;
+  if (!capture || capture.runId || !context.runId) return;
+  if (typeof event.prompt !== "string" || !event.prompt.includes(capture.promptMarker)) return;
+  capture.runId = context.runId;
+}
+
+export function recordA2AProgressToolActivity(
+  event: ToolHookEvent,
+  context: HookContext,
+): void {
+  const capture = context.sessionKey ? captures.get(context.sessionKey) : undefined;
+  const runId = event.runId ?? context.runId;
+  if (!capture?.runId || capture.runId !== runId || !event.toolName) return;
+  const next = normalizeA2AToolIdentifier(event.toolName);
+  if (!next || capture.toolIdentifiers.at(-1) === next) return;
+  capture.toolIdentifiers.push(next);
+  if (capture.toolIdentifiers.length > MAX_TOOL_IDENTIFIERS) {
+    capture.toolIdentifiers.shift();
+  }
+}

src/a2a-progress.ts

@@ -0,0 +1,96 @@
+import {
+  normalizeA2AIdentifierText,
+  normalizeA2AToolIdentifier,
+} from "./a2a-progress-activity.js";
+
+export const DEFAULT_A2A_PROGRESS_INTERVAL_SECONDS = 180;
+
+const TERMINAL_CLAIM_RE =
+  /\b(?:done|complete|completed|finished|failed|failure|blocked|final\s+(?:answer|result)|cannot\s+(?:complete|continue)|need(?:ed|s)?\s+(?:your\s+)?input|waiting\s+(?:for\s+)?(?:your\s+)?input|waiting\s+for\s+you)\b/i;
+
+export function resolveA2AProgressIntervalSeconds(value: unknown): number {
+  return typeof value === "number" && Number.isFinite(value) && value >= 0
+    ? Math.floor(value)
+    : DEFAULT_A2A_PROGRESS_INTERVAL_SECONDS;
+}
+
+export function a2aReceiptText(taskId: string, intervalSeconds: number): string {
+  if (intervalSeconds <= 0) {
+    return `Task ${taskId} received. Work is queued and starting. Periodic progress updates are disabled.`;
+  }
+  const cadence = intervalSeconds % 60 === 0
+    ? `about every ${intervalSeconds / 60} ${intervalSeconds === 60 ? "minute" : "minutes"}`
+    : `about every ${intervalSeconds} seconds`;
+  return `Task ${taskId} received. Work is queued and starting. Expect progress updates ${cadence}.`;
+}
+
+export function a2aProgressFallback(elapsedSeconds: number): string {
+  return `I'm continuing the requested work. (${elapsedSeconds}s elapsed)`;
+}
+
+export function sanitizeA2AProgressText(
+  value: string,
+  toolIdentifiers: string[],
+  elapsedSeconds: number,
+): string {
+  const fallback = a2aProgressFallback(elapsedSeconds);
+  const normalized = value.replace(/\s+/g, " ").trim();
+  const withoutElapsed = normalized.replace(/\s*\(\d+s elapsed\)\s*$/i, "").trim();
+  const normalizedText = normalizeA2AIdentifierText(withoutElapsed);
+  const repeatsIdentifier = toolIdentifiers.some((identifier) => {
+    const safeIdentifier = normalizeA2AToolIdentifier(identifier);
+    return safeIdentifier.length > 0 && new RegExp(
+      `(?:^|_)${safeIdentifier.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}(?:_|$)`,
+    ).test(normalizedText);
+  });
+  if (
+    !withoutElapsed ||
+    repeatsIdentifier ||
+    TERMINAL_CLAIM_RE.test(withoutElapsed)
+  ) {
+    return fallback;
+  }
+  const words = withoutElapsed.split(" ").slice(0, 16).join(" ").slice(0, 180).trim();
+  if (!words) return fallback;
+  return `${words.replace(/[.!?]+$/, "")}. (${elapsedSeconds}s elapsed)`;
+}
+
+export function taskAgentHistoryContains(task: unknown, expected: string): boolean {
+  const seen = new Set<unknown>();
+  const visit = (value: unknown): boolean => {
+    if (value === expected) return true;
+    if (!value || typeof value !== "object" || seen.has(value)) return false;
+    seen.add(value);
+    if (Array.isArray(value)) return value.some(visit);
+    return Object.values(value as Record<string, unknown>).some(visit);
+  };
+  if (!task || typeof task !== "object") return false;
+  const record = task as {
+    messages?: unknown;
+    raw?: { history?: unknown };
+  };
+  const messages = Array.isArray(record.messages)
+    ? record.messages
+    : Array.isArray(record.raw?.history)
+      ? record.raw.history
+      : [];
+  return messages.some((message) => {
+    if (!message || typeof message !== "object") return false;
+    const entry = message as { role?: unknown; parts?: unknown };
+    const role = String(entry.role ?? "").toLowerCase();
+    return (role === "agent" || role === "role_agent") && visit(entry.parts);
+  });
+}
+
+export function abortableDelay(milliseconds: number, signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return Promise.resolve();
+  return new Promise((resolve) => {
+    const timer = setTimeout(finish, milliseconds);
+    function finish() {
+      signal.removeEventListener("abort", finish);
+      clearTimeout(timer);
+      resolve();
+    }
+    signal.addEventListener("abort", finish, { once: true });
+  });
+}