[VascoSch92]

HUMAN:

Fixing a bug where we were publishing every delta instead of just the event.

---

AGENT:

Why

StreamingDeltaEvent is published straight to the conversation PubSub (event_service.py:1061), and PubSub.__call__ notified every registered subscriber with no kind filter. Making the delta an Event opted it into every consumer of the bus by default, so consumers sized for conversation events started receiving traffic proportional to tokens:

• [Bug]: every token delta is POSTed to configured webhooks #4672 — WebhookSubscriber.__call__ enqueues unconditionally, and event_buffer_size defaults to 5, so a streamed response became a continuous POST loop to the configured endpoint for its whole duration. Model output is not masked on the standard Agent path, so that text was leaving the process one fragment at a time.
• [Bug]: telemetry event_count_bucket counts token deltas, making it incomparable across conversations #4673 — TelemetrySubscriber._handle increments _event_count before its match, so event_count_bucket scaled with model verbosity instead of conversation activity, making streamed and non-streamed conversations incomparable.

Same root cause, so one fix closes both.

Summary

• Add Subscriber.receives_streaming_deltas (ClassVar, defaults to False) and consult it in PubSub.__call__: deltas are delivered only to subscribers that opt in.
• _WebSocketSubscriber opts in — the live socket is the one consumer token streaming exists for.
• Regression tests for both acceptance criteria, plus one for the central bus behaviour.

Delivery is opt-in rather than opt-out so a subscriber added later cannot inherit token-rate traffic — and the content it carries — by accident. This is option 2 from #4672 ("interim: a kind filter"); it does not preclude the epic's structural direction of taking deltas off the bus entirely.

Issue Number

Fixes #4672
Fixes #4673

How to Test

End-to-end — real HTTP egress. The script below runs a real HTTP receiver, a real PubSub and a real WebhookSubscriber, then publishes one streamed response (3 conversation events + 200 token deltas) with the production default event_buffer_size=5. It uses no new API, so it runs unchanged on main:

webhook_delta_e2e.py

import asyncio, json, threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from unittest.mock import MagicMock
from uuid import uuid4

from openhands.agent_server.config import WebhookSpec
from openhands.agent_server.conversation_service import WebhookSubscriber
from openhands.agent_server.pub_sub import PubSub
from openhands.sdk.event import StreamingDeltaEvent
from openhands.sdk.event.llm_convertible import MessageEvent
from openhands.sdk.llm.message import Message, TextContent

POSTS: list[list[dict]] = []

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        n = int(self.headers.get("Content-Length", 0))
        POSTS.append(json.loads(self.rfile.read(n)))
        self.send_response(200); self.end_headers(); self.wfile.write(b"{}")
    def log_message(self, *a): pass

async def main():
    server = HTTPServer(("127.0.0.1", 0), Handler)
    port = server.server_address[1]
    threading.Thread(target=server.serve_forever, daemon=True).start()

    subscriber = WebhookSubscriber(
        conversation_id=uuid4(), service=MagicMock(),
        spec=WebhookSpec(base_url=f"http://127.0.0.1:{port}",
                         event_buffer_size=5, num_retries=0, flush_delay=0.05),
    )
    pub_sub = PubSub()
    pub_sub.subscribe(subscriber)

    def msg(i):
        return MessageEvent(source="user",
            llm_message=Message(role="user", content=[TextContent(text=f"turn {i}")]))

    n_conv, n_delta = 3, 200
    for i in range(n_conv):
        await pub_sub(msg(i))
        for _ in range(n_delta // n_conv):
            await pub_sub(StreamingDeltaEvent(content="tok "))

    await asyncio.sleep(0.4); await subscriber.close(); await asyncio.sleep(0.2)
    server.shutdown()

    delivered = [e for batch in POSTS for e in batch]
    deltas = [e for e in delivered if e.get("kind") == "StreamingDeltaEvent"]
    print(f"  HTTP POSTs sent  : {len(POSTS)}")
    print(f"  events delivered : {len(delivered)}")
    print(f"  DELTAS DELIVERED : {len(deltas)}")

asyncio.run(main())

$ uv run python webhook_delta_e2e.py     # on main (before)
  published        : 3 conversation events + ~200 deltas
  HTTP POSTs sent  : 41
  events delivered : 201
  DELTAS DELIVERED : 198

$ uv run python webhook_delta_e2e.py     # on this branch (after)
  published        : 3 conversation events + ~200 deltas
  HTTP POSTs sent  : 1
  events delivered : 3
  DELTAS DELIVERED : 0

41 POSTs → 1, and that single POST is exactly what the equivalent non-streamed response sends — which is acceptance criterion 1 of #4672.

Unit tests. Each fails on main and passes here:

$ uv run pytest \
    tests/agent_server/test_event_streaming.py::test_deltas_only_reach_subscribers_that_opted_in \
    tests/agent_server/test_webhook_subscriber.py::test_streaming_deltas_never_reach_webhook \
    tests/agent_server/telemetry/test_telemetry_subscriber.py::test_event_count_ignores_streaming_deltas -q

# on main:        3 failed
#   queue == []            -> AssertionError: Left contains 5 more items ...
#   sub._event_count == 0  -> AssertionError: assert 50 == 0
# on this branch: 3 passed

Regression sweep. Full agent-server suite plus pre-commit:

$ uv run pytest tests/agent_server/ -q
  2073 passed, 13 deselected in 269.61s

$ uv run pre-commit run --files <changed files>
  Ruff format / Ruff lint / pycodestyle / pyright /
  Check import dependency rules / Check Tool subclass registration ... Passed

Video/Screenshots

Terminal output inline above — the before/after POST counts are the observable behaviour of this change.

Design Doc

n/a — the change is 11 added lines of source across two files.

Type

• Bug fix
• Feature
• Refactor
• Breaking change
• Docs / chore

Notes

• Metric discontinuity ([Bug]: telemetry event_count_bucket counts token deltas, making it incomparable across conversations #4673). event_count_bucket will step down for streamed conversations once this lands. That is the correction, but it is still a break in the series — worth a heads-up to whoever consumes the dashboard rather than a silent change.
• Why opt-in and not two isinstance guards. [Bug]: every token delta is POSTed to configured webhooks #4672 offers either. Two guards fix today's two subscribers; the flag also protects the next one. There are currently five subscribers on this bus (_EventSubscriber, AutoTitleSubscriber, WebhookSubscriber, TelemetrySubscriber, _WebSocketSubscriber) and only the last wants deltas.
• Idle eviction is unaffected. _EventSubscriber calls service.touch() on every event, so it no longer sees deltas — but is_idle_evictable() already returns False whenever _run_task is active, and a streamed response is by definition an active run. Dropping deltas there also removes a stored.updated_at write per token.
• PubSub stays generic. The isinstance check is a no-op for PubSub[BashEventBase]; the pre-commit import-dependency rule passes.
• Does not close the epic's structural item (deltas off the bus entirely) or the unmasked-model-output issue [Security] Model output is never masked on the standard Agent path #4678, which is a separate path.

---

🐳 Agent Server images for this PR — GHCR package, pull/run commands, and all pushed tags (click to expand)

• GHCR package: https://github.com/OpenHands/agent-sdk/pkgs/container/agent-server

Variants & Base Images

Variant	Architectures	Base Image	Docs / Tags
java	amd64, arm64	eclipse-temurin:17-jdk	Link
python	amd64, arm64	nikolaik/python-nodejs:python3.13-nodejs22-slim	Link
golang	amd64, arm64	golang:1.21-bookworm	Link

Pull (multi-arch manifest)

# Each variant is a multi-arch manifest supporting both amd64 and arm64
docker pull ghcr.io/openhands/agent-server:1f2aef9-python

Run

docker run -it --rm \
  -p 8000:8000 \
  --name agent-server-1f2aef9-python \
  ghcr.io/openhands/agent-server:1f2aef9-python

All tags pushed for this build

ghcr.io/openhands/agent-server:1f2aef9-golang-amd64
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-golang-amd64
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-golang-amd64
ghcr.io/openhands/agent-server:1f2aef9-golang_tag_1.21-bookworm-amd64
ghcr.io/openhands/agent-server:1f2aef9-golang-arm64
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-golang-arm64
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-golang-arm64
ghcr.io/openhands/agent-server:1f2aef9-golang_tag_1.21-bookworm-arm64
ghcr.io/openhands/agent-server:1f2aef9-java-amd64
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-java-amd64
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-java-amd64
ghcr.io/openhands/agent-server:1f2aef9-eclipse-temurin_tag_17-jdk-amd64
ghcr.io/openhands/agent-server:1f2aef9-java-arm64
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-java-arm64
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-java-arm64
ghcr.io/openhands/agent-server:1f2aef9-eclipse-temurin_tag_17-jdk-arm64
ghcr.io/openhands/agent-server:1f2aef9-python-amd64
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-python-amd64
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-python-amd64
ghcr.io/openhands/agent-server:1f2aef9-nikolaik_s_python-nodejs_tag_python3.13-nodejs22-slim-amd64
ghcr.io/openhands/agent-server:1f2aef9-python-arm64
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-python-arm64
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-python-arm64
ghcr.io/openhands/agent-server:1f2aef9-nikolaik_s_python-nodejs_tag_python3.13-nodejs22-slim-arm64
ghcr.io/openhands/agent-server:1f2aef9-golang
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-golang
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-golang
ghcr.io/openhands/agent-server:1f2aef9-golang_tag_1.21-bookworm
ghcr.io/openhands/agent-server:1f2aef9-java
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-java
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-java
ghcr.io/openhands/agent-server:1f2aef9-eclipse-temurin_tag_17-jdk
ghcr.io/openhands/agent-server:1f2aef9-python
ghcr.io/openhands/agent-server:1f2aef96cf7f7f20dacb79bf887a75c4792a6ed3-python
ghcr.io/openhands/agent-server:vasco-filter-streaming-deltas-from-subscribers-python
ghcr.io/openhands/agent-server:1f2aef9-nikolaik_s_python-nodejs_tag_python3.13-nodejs22-slim

About Multi-Architecture Support

• Each variant tag (e.g., 1f2aef9-python) is a multi-arch manifest supporting both amd64 and arm64
• Docker automatically pulls the correct architecture for your platform
• Individual architecture tags (e.g., 1f2aef9-python-amd64) are also available if needed

[github-actions]

Python API breakage checks — ✅ PASSED

Result: ✅ PASSED

Action log

[github-actions]

REST API breakage checks (OpenAPI) — ✅ PASSED

Result: ✅ PASSED

Action log

[github-actions]

Coverage Report •

File	Stmts	Miss	Cover	Missing
openhands-agent-server/openhands/agent_server
pub_sub.py	50	2	96%	102–103
sockets.py	243	26	89%	202, 205, 208, 286–287, 291, 293–295, 301–303, 306, 427–430, 435–436, 441–443, 513, 515, 534, 594
TOTAL	41502	17271	58%

[enyst]

As noted on the fly in the first issue, I think it’s worth thinking about making deltas non-Events, that is, remove the inheritance. I’m not sure off hand what the consequences are, it’s just, it’s been in the back of my mind since your HTML design doc.

Sorry, I know it’s not exactly in scope in this PR itself. Just me in a hurry. 🙏