feat: emit explicit interrupt resolved status updates

docs/guide.md

@@ -72,9 +72,13 @@ Key variables to understand protocol behavior:
   such as `step-finish`; non-usage parts with similar fields are ignored.
   Interrupt events (`permission.asked` / `question.asked`) are mapped to
   `TaskStatusUpdateEvent(final=false, state=input-required)` with details at
-  `metadata.shared.interrupt` (including `request_id`, interrupt `type`, and
-  normalized minimal callback payload). Resolved interrupt events only clear
-  internal pending state; they do not emit a separate outward status event.
+  `metadata.shared.interrupt` (including `request_id`, interrupt `type`,
+  `phase=asked`, and normalized minimal callback payload). Resolved interrupt
+  events (`permission.replied` / `question.replied` / `question.rejected`) are
+  emitted as `TaskStatusUpdateEvent(final=false, state=working)` with
+  `metadata.shared.interrupt.phase=resolved` and a normalized
+  `metadata.shared.interrupt.resolution`. Duplicate or unknown resolved events
+  are suppressed unless the matching request is still pending.
   Non-streaming requests return a `Task` directly.
 - Non-streaming `message:send` responses may include normalized token usage at
   `Task.metadata.shared.usage` with the same field schema.

src/opencode_a2a_serve/agent.py

@@ -157,8 +157,14 @@ def mark_interrupt_pending(self, request_id: str) -> bool:
         self.pending_interrupt_request_ids.add(normalized)
         return True
 
-    def clear_interrupt_pending(self, request_id: str) -> None:
-        self.pending_interrupt_request_ids.discard(request_id.strip())
+    def clear_interrupt_pending(self, request_id: str) -> bool:
+        normalized = request_id.strip()
+        if not normalized:
+            return False
+        if normalized not in self.pending_interrupt_request_ids:
+            return False
+        self.pending_interrupt_request_ids.discard(normalized)
+        return True
 
 
 @dataclass(frozen=True)
@@ -1097,8 +1103,19 @@ async def _emit_interrupt_status(
             state: TaskState,
             request_id: str,
             interrupt_type: str,
-            details: Mapping[str, Any],
+            phase: str,
+            details: Mapping[str, Any] | None = None,
+            resolution: str | None = None,
         ) -> None:
+            interrupt_metadata: dict[str, Any] = {
+                "request_id": request_id,
+                "type": interrupt_type,
+                "phase": phase,
+            }
+            if details is not None:
+                interrupt_metadata["details"] = dict(details)
+            if resolution is not None:
+                interrupt_metadata["resolution"] = resolution
             sequence = stream_state.next_sequence()
             await event_queue.enqueue_event(
                 TaskStatusUpdateEvent(
@@ -1115,9 +1132,7 @@ async def _emit_interrupt_status(
                             "sequence": sequence,
                         },
                         interrupt={
-                            "request_id": request_id,
-                            "type": interrupt_type,
-                            "details": dict(details),
+                            **interrupt_metadata,
                         },
                     ),
                 )
@@ -1326,17 +1341,28 @@ def _tool_chunks(
                                         state=TaskState.input_required,
                                         request_id=request_id,
                                         interrupt_type=asked["interrupt_type"],
+                                        phase="asked",
                                         details=asked["details"],
                                     )
                             resolved = _extract_interrupt_resolved_event(event)
                             if resolved is not None:
                                 resolved_request_id = resolved["request_id"]
-                                stream_state.clear_interrupt_pending(resolved_request_id)
+                                cleared_pending = stream_state.clear_interrupt_pending(
+                                    resolved_request_id
+                                )
                                 discard_request = getattr(
                                     self._client, "discard_interrupt_request", None
                                 )
                                 if callable(discard_request):
                                     discard_request(resolved_request_id)
+                                if cleared_pending:
+                                    await _emit_interrupt_status(
+                                        state=TaskState.working,
+                                        request_id=resolved_request_id,
+                                        interrupt_type=resolved["interrupt_type"],
+                                        phase="resolved",
+                                        resolution=resolved["resolution"],
+                                    )
                         if event_type not in {"message.part.updated", "message.part.delta"}:
                             continue
                         part = props.get("part")
@@ -1842,7 +1868,14 @@ def _extract_interrupt_resolved_event(event: Mapping[str, Any]) -> dict[str, str
     request_id = _extract_interrupt_resolved_request_id(props)
     if not request_id:
         return None
-    return {"request_id": request_id, "event_type": event_type}
+    interrupt_type = "permission" if event_type.startswith("permission.") else "question"
+    resolution = "rejected" if event_type == "question.rejected" else "replied"
+    return {
+        "request_id": request_id,
+        "event_type": event_type,
+        "interrupt_type": interrupt_type,
+        "resolution": resolution,
+    }
 
 
 def _extract_stream_message_id(part: Mapping[str, Any], props: Mapping[str, Any]) -> str | None:

tests/test_streaming_output_contract.py

@@ -727,6 +727,7 @@ async def test_streaming_emits_interrupt_status_for_permission_asked_event() ->
     assert len(interrupt_statuses) == 1
     interrupt = _interrupt_meta(interrupt_statuses[0])
     assert interrupt["request_id"] == "perm-req-1"
+    assert interrupt["phase"] == "asked"
     assert interrupt["details"]["permission"] == "read"
     assert "/data/project/.env.secret" in interrupt["details"]["patterns"]
     assert "metadata" not in interrupt["details"]
@@ -762,6 +763,7 @@ async def test_streaming_emits_interrupt_status_for_question_asked_event() -> No
     assert len(interrupt_statuses) == 1
     interrupt = _interrupt_meta(interrupt_statuses[0])
     assert interrupt["request_id"] == "q-req-1"
+    assert interrupt["phase"] == "asked"
     assert interrupt["details"]["questions"] == [
         {
             "header": "Confirm",
@@ -885,14 +887,71 @@ async def test_streaming_resolved_interrupt_only_clears_internal_pending_state()
         and event.final is False
         and (event.metadata or {}).get("shared", {}).get("interrupt") is not None
     ]
-    assert len(interrupt_statuses) == 1
-    assert _interrupt_meta(interrupt_statuses[0])["request_id"] == "perm-req-resolve"
+    assert len(interrupt_statuses) == 2
+    asked_interrupt = _interrupt_meta(interrupt_statuses[0])
+    resolved_interrupt = _interrupt_meta(interrupt_statuses[1])
+    assert asked_interrupt["request_id"] == "perm-req-resolve"
+    assert asked_interrupt["phase"] == "asked"
+    assert interrupt_statuses[0].status.state == TaskState.input_required
+    assert resolved_interrupt["request_id"] == "perm-req-resolve"
+    assert resolved_interrupt["type"] == "permission"
+    assert resolved_interrupt["phase"] == "resolved"
+    assert resolved_interrupt["resolution"] == "replied"
+    assert "details" not in resolved_interrupt
+    assert interrupt_statuses[1].status.state == TaskState.working
     final_status = [
         event for event in queue.events if isinstance(event, TaskStatusUpdateEvent) and event.final
     ][-1]
     assert "interrupt" not in (final_status.metadata or {}).get("shared", {})
 
 
+@pytest.mark.asyncio
+async def test_streaming_duplicate_interrupt_resolved_event_is_not_emitted_twice() -> None:
+    client = DummyStreamingClient(
+        stream_events_payload=[
+            _question_asked_event(session_id="ses-1", request_id="q-req-resolve"),
+            _interrupt_resolved_event(
+                session_id="ses-1",
+                request_id="q-req-resolve",
+                event_type="question.rejected",
+            ),
+            _interrupt_resolved_event(
+                session_id="ses-1",
+                request_id="q-req-resolve",
+                event_type="question.rejected",
+            ),
+            _event(session_id="ses-1", role="assistant", part_type="text", delta="answer"),
+        ],
+        response_text="answer",
+    )
+    executor = OpencodeAgentExecutor(client, streaming_enabled=True)
+    executor._should_stream = lambda context: True  # type: ignore[method-assign]
+    queue = DummyEventQueue()
+
+    await executor.execute(
+        make_request_context(
+            task_id="task-interrupt-resolved-dedupe",
+            context_id="ctx-interrupt-resolved-dedupe",
+            text="hello",
+        ),
+        queue,
+    )
+
+    interrupt_statuses = [
+        event
+        for event in queue.events
+        if isinstance(event, TaskStatusUpdateEvent)
+        and event.final is False
+        and (event.metadata or {}).get("shared", {}).get("interrupt") is not None
+    ]
+    assert len(interrupt_statuses) == 2
+    assert _interrupt_meta(interrupt_statuses[0])["phase"] == "asked"
+    resolved_interrupt = _interrupt_meta(interrupt_statuses[1])
+    assert resolved_interrupt["phase"] == "resolved"
+    assert resolved_interrupt["type"] == "question"
+    assert resolved_interrupt["resolution"] == "rejected"
+
+
 def _unique(items: list[str]) -> list[str]:
     seen: set[str] = set()
     ordered: list[str] = []