mirror of
https://github.com/andrewyng/openworker.git
synced 2026-11-04 13:27:11 +00:00
Devika Verma
c1e8b8cdf3
The per-turn `<system-context>` block opened with `Now: YYYY-MM-DD HH:MM` (2c9fb74, 2026-08-20) so that absolute scheduling had a clock. The block is glued onto the last user message — the task prompt in a tool loop — and a provider's prompt cache is reusable only up to the first byte that differs, so every minute crossing rewrote message one and re-processed the whole conversation. 9d6a716 worked around it with a trailing-message shape, an Anthropic-only relocation of that message, and a `live_clock` switch that headless runs set off. This commit removes the cause instead and deletes the workarounds; the shape is the same on every provider again. - The block carries no clock, on any provider. What is left (folders, skill menu, mode notices, memory-off notice, disable countermand) changes only when the user changes something, so the outbound history is byte-identical turn to turn. - `current_time` tool (coworker/clock.py), registered for every persona: local time with offset and zone, the same instant in UTC, weekday. The environment block's "Today's date" now says it is a session-start snapshot and points at the tool. - `sleep_for(seconds)` is back beside `sleep_until` (a relative wait needs no clock arithmetic; the four lead personas' manifests still told the model to call sleep_for, which2c9fb74had removed). `sleep_until`'s description points at current_time for the date/timezone. Timer wake messages carry their fire time. - Deleted: `live_clock` config + COWORKER_LIVE_CLOCK, the trailing-message shape in `_outbound_messages`, `_relocate_ephemeral_note` in the Anthropic provider. The cache-diagnostics instrument (off by default) stays. Validation runs (one long tool-heavy task, one attempt each on Fable 5.1 and Kimi K3, compared with the 9d6a716 runs of the same task) are recorded on OPE-192. Tests: tests/test_clock_tool.py (new), tests/test_context_block_placement.py and tests/test_cache_breakpoint_placement.py rewritten for the single shape, tests/test_live_clock.py removed; self-wake, attachments, message-source and multiroot tests updated. Full suite unchanged against the Windows baseline.
263 lines
9.3 KiB
Python
263 lines
9.3 KiB
Python
"""Phase 2 — structured connector inbound messages (UI-REFRESH §3).
|
|
|
|
A connector message reaches a session as a framed text blob (model-facing `content`) plus a
|
|
display-only `MessageSource` sidecar. The sidecar is persisted (so `GET /messages` renders a card)
|
|
but stripped before any message reaches a provider. These tests pin that contract.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from coworker.connectors.base import MessageEvent, MessageSource, SessionSource
|
|
from coworker.engine import TurnEngine
|
|
from coworker.permissions import PermissionEngine
|
|
from coworker.providers import AssistantTurn, ModelCapabilities, ProviderClient
|
|
from coworker.server import create_app
|
|
from coworker.server.manager import SessionManager
|
|
from coworker.tools import ToolRegistry
|
|
|
|
|
|
class CapturingProvider(ProviderClient):
|
|
"""Returns queued turns and records the messages handed to it on each call — so a test can
|
|
assert exactly what the provider saw (framed text, no `source`)."""
|
|
|
|
def __init__(self, turns):
|
|
self._turns = list(turns)
|
|
self.calls: list[list[dict]] = []
|
|
|
|
def complete(self, *, model, messages, tools=None, **settings):
|
|
self.calls.append([dict(m) for m in messages])
|
|
return self._turns.pop(0)
|
|
|
|
def capabilities(self, model):
|
|
return ModelCapabilities()
|
|
|
|
|
|
def _text(text):
|
|
return AssistantTurn(text=text, finish_reason="stop")
|
|
|
|
|
|
def _channel_event(text="deploy failed", chat_id="C1"):
|
|
return MessageEvent(
|
|
text=text,
|
|
source=SessionSource(
|
|
platform="slack",
|
|
chat_id=chat_id,
|
|
user_id="U1",
|
|
user_name="Bob",
|
|
chat_name="#ocw-test",
|
|
chat_type="channel",
|
|
),
|
|
message_id="1700000001.000001",
|
|
)
|
|
|
|
|
|
def _dm_event(text="ping me", chat_id="D1"):
|
|
return MessageEvent(
|
|
text=text,
|
|
source=SessionSource(
|
|
platform="slack",
|
|
chat_id=chat_id,
|
|
user_id="U2",
|
|
user_name="Sue",
|
|
chat_name="Sue",
|
|
chat_type="dm",
|
|
),
|
|
message_id="1700000002.000002",
|
|
)
|
|
|
|
|
|
def _connect_slack(mgr):
|
|
"""Inbound delivery is gated on the connector being CONNECTED (§4.3). Tests used to pass
|
|
by riding the developer's real Slack profile; with the isolated state dir (conftest) each
|
|
test must connect its own."""
|
|
mgr.secrets.put(
|
|
"slack:default",
|
|
{"bot_token": "xoxb-test", "app_token": "xapp-test", "enabled": True},
|
|
)
|
|
|
|
|
|
def test_inbound_builds_message_source(tmp_path, monkeypatch):
|
|
mgr = SessionManager(workspace=tmp_path, provider=CapturingProvider([]))
|
|
_connect_slack(mgr)
|
|
captured: list[tuple] = []
|
|
|
|
async def fake_deliver(session_id, message, *, source=None):
|
|
captured.append((session_id, message, source))
|
|
|
|
monkeypatch.setattr(mgr, "deliver_to_session", fake_deliver)
|
|
mgr.subscriptions.subscribe("sA", "slack:C1")
|
|
|
|
asyncio.run(mgr._dispatch_inbound(_channel_event()))
|
|
|
|
assert len(captured) == 1
|
|
sid, message, source = captured[0]
|
|
assert sid == "sA"
|
|
assert source == {
|
|
"connector": "slack",
|
|
"kind": "channel",
|
|
"channel_id": "C1",
|
|
"channel_name": "#ocw-test",
|
|
"sender_id": "U1",
|
|
"sender_name": "Bob",
|
|
"ts": float("1700000001.000001"),
|
|
"text": "deploy failed",
|
|
}
|
|
# the delivered `message` is the FRAMED (model-facing) text, not the raw message
|
|
assert "subscribed" in message and "deploy failed" in message
|
|
|
|
|
|
def test_message_source_persisted_and_stripped(tmp_path):
|
|
provider = CapturingProvider([_text("ack")])
|
|
mgr = SessionManager(workspace=tmp_path, provider=provider)
|
|
_connect_slack(mgr)
|
|
mgr.get_engine("S", agent="chat") # durable, workspace-free session
|
|
mgr.subscriptions.subscribe("S", "slack:C1")
|
|
|
|
asyncio.run(mgr._dispatch_inbound(_channel_event()))
|
|
|
|
client = TestClient(create_app(mgr))
|
|
messages = client.get("/v1/sessions/S/messages").json()["messages"]
|
|
user_msgs = [m for m in messages if m.get("role") == "user"]
|
|
assert user_msgs, messages
|
|
last_user = user_msgs[-1]
|
|
|
|
# persisted WITH the display-only source sidecar
|
|
assert last_user["source"]["connector"] == "slack"
|
|
assert last_user["source"]["channel_name"] == "#ocw-test"
|
|
assert last_user["source"]["sender_name"] == "Bob"
|
|
assert last_user["source"]["text"] == "deploy failed" # raw message on the card
|
|
# the model-facing content stays the FRAMED text (raw message is NOT the content)
|
|
assert "subscribed" in last_user["content"]
|
|
assert last_user["content"] != "deploy failed"
|
|
|
|
# the provider was called and saw the framed text with NO source / unknown keys
|
|
assert provider.calls, "provider should have been invoked"
|
|
sent_user = [m for m in provider.calls[0] if m.get("role") == "user"][-1]
|
|
assert "source" not in sent_user
|
|
assert "subscribed" in sent_user["content"] # framed, not raw
|
|
# NO message handed to ANY provider call carries a source key
|
|
assert all("source" not in m for call in provider.calls for m in call)
|
|
|
|
|
|
def test_outbound_strips_source_with_and_without_context(tmp_path):
|
|
"""Direct engine check that the `source` strip is UNCONDITIONAL — it happens on the
|
|
no-context early-return path too, not only when a `<system-context>` block is added.
|
|
"""
|
|
registry = ToolRegistry()
|
|
permissions = PermissionEngine(workspace_root=tmp_path)
|
|
src = {"connector": "slack", "kind": "channel", "text": "raw"}
|
|
|
|
# no context provider → the early-return path must still strip `source`
|
|
engine = TurnEngine(
|
|
provider=CapturingProvider([]),
|
|
registry=registry,
|
|
permissions=permissions,
|
|
model="gpt-5.5",
|
|
)
|
|
engine.messages.append({"role": "user", "content": "framed", "source": src})
|
|
out = engine._outbound_messages()
|
|
assert all("source" not in m for m in out)
|
|
# self.messages is never mutated — the sidecar is preserved for persistence
|
|
assert engine.messages[-1]["source"] == src
|
|
|
|
# with a context provider the strip still holds, plus the block is appended
|
|
engine2 = TurnEngine(
|
|
provider=CapturingProvider([]),
|
|
registry=registry,
|
|
permissions=permissions,
|
|
model="gpt-5.5",
|
|
context_provider=lambda: "live ctx",
|
|
)
|
|
engine2.messages.append({"role": "user", "content": "framed", "source": src})
|
|
out2 = engine2._outbound_messages()
|
|
assert all("source" not in m for m in out2)
|
|
assert "<system-context>" in out2[-1]["content"]
|
|
assert engine2.messages[-1]["source"] == src # original untouched
|
|
|
|
|
|
def test_tool_display_sidecar_is_agent_invisible(tmp_path):
|
|
"""`_display` on a tool result (e.g. gmail filter-hidden counts) mirrors the
|
|
`source` contract: lifted onto the message for the GUI, audited as a rule+count
|
|
row, and stripped from every provider feed — the agent sees no tombstone."""
|
|
from coworker.providers.base import ToolCall
|
|
|
|
audits: list[dict] = []
|
|
engine = TurnEngine(
|
|
provider=CapturingProvider([]),
|
|
registry=ToolRegistry(),
|
|
permissions=PermissionEngine(workspace_root=tmp_path),
|
|
model="gpt-5.5",
|
|
audit_sink=audits.append,
|
|
)
|
|
tc = ToolCall(id="t1", name="gmail_search_messages", arguments={"query": "q"})
|
|
engine._record_result(
|
|
tc,
|
|
{
|
|
"ok": True,
|
|
"data": {"messages": [{"id": "m2"}]},
|
|
"_display": {"hidden_by_filters": 2, "connector": "gmail"},
|
|
},
|
|
"ok",
|
|
)
|
|
|
|
msg = engine.messages[-1]
|
|
assert msg["_display"] == {"hidden_by_filters": 2, "connector": "gmail"}
|
|
assert "_display" not in msg["content"] and "hidden" not in msg["content"]
|
|
|
|
out = engine._outbound_messages()
|
|
assert all("_display" not in m for m in out)
|
|
assert engine.messages[-1]["_display"] # persisted for the tool card
|
|
|
|
filtered = [a for a in audits if a.get("stage") == "filtered"]
|
|
assert len(filtered) == 1
|
|
assert "2 result(s) hidden" in filtered[0]["reason"]
|
|
|
|
|
|
def test_turn_start_carries_source(tmp_path):
|
|
mgr = SessionManager(workspace=tmp_path, provider=CapturingProvider([_text("ok")]))
|
|
_connect_slack(mgr)
|
|
mgr.get_engine("S", agent="chat")
|
|
mgr.subscriptions.subscribe("S", "slack:C1")
|
|
|
|
events: list[dict] = []
|
|
|
|
async def cb(msg):
|
|
events.append(msg)
|
|
|
|
mgr.register_session_client("S", cb)
|
|
|
|
asyncio.run(mgr._dispatch_inbound(_channel_event()))
|
|
|
|
starts = [e for e in events if e["type"] == "turn_start"]
|
|
assert starts, events
|
|
source = starts[0]["data"]["source"]
|
|
assert source["connector"] == "slack"
|
|
assert source["kind"] == "channel"
|
|
assert source["channel_name"] == "#ocw-test"
|
|
assert source["text"] == "deploy failed"
|
|
|
|
|
|
def test_dm_message_source_kind_dm(tmp_path, monkeypatch):
|
|
mgr = SessionManager(workspace=tmp_path, provider=CapturingProvider([]))
|
|
_connect_slack(mgr)
|
|
captured: list[tuple] = []
|
|
|
|
async def fake_deliver(session_id, message, *, source=None):
|
|
captured.append((session_id, message, source))
|
|
|
|
monkeypatch.setattr(mgr, "deliver_to_session", fake_deliver)
|
|
mgr.set_dm_session("sDM")
|
|
|
|
asyncio.run(mgr._dispatch_inbound(_dm_event()))
|
|
|
|
assert len(captured) == 1
|
|
sid, message, source = captured[0]
|
|
assert sid == "sDM"
|
|
assert source["kind"] == "dm"
|
|
assert source["connector"] == "slack"
|
|
assert source["channel_id"] == "D1"
|
|
assert source["sender_name"] == "Sue"
|
|
assert source["text"] == "ping me"
|