278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""Relay Slack live cards — gateway half (NS-658).
|
|
|
|
The relay adapter emits three new additive ops when the connector's
|
|
negotiated descriptor advertises them:
|
|
|
|
{op: "draft", chat_id, draft_id, content, final, metadata}
|
|
{op: "task_card", chat_id, card_id, chunks, metadata}
|
|
{op: "task_card_stop", chat_id, card_id, metadata}
|
|
|
|
Gateway side is deliberately dumb: no Slack API knowledge, no new config
|
|
keys. Capability is descriptor-driven — an old connector that never
|
|
advertises "draft"/"task_card" gets byte-identical behavior to today.
|
|
|
|
Semantic bridge: the base send_draft contract is Telegram-shaped (draft
|
|
clears; final answer is a separate send). Slack native streaming makes the
|
|
stream THE message. The adapter tracks the open draft per chat; the
|
|
turn-final regular send() for that chat converts to draft(final=true) so
|
|
the connector seals the stream instead of posting a duplicate.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from gateway.config import PlatformConfig
|
|
from gateway.relay.adapter import RelayAdapter
|
|
from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
|
|
from tests.gateway.relay.stub_connector import StubConnector
|
|
|
|
|
|
def make_desc(**kw) -> CapabilityDescriptor:
|
|
base = dict(
|
|
contract_version=CONTRACT_VERSION,
|
|
platform="slack",
|
|
label="Slack",
|
|
max_message_length=39000,
|
|
supports_draft_streaming=True,
|
|
supports_edit=True,
|
|
supports_threads=True,
|
|
markdown_dialect="slack",
|
|
len_unit="chars",
|
|
emoji="\U0001f4ac",
|
|
platform_hint="",
|
|
pii_safe=False,
|
|
supported_ops=("send", "edit", "typing", "draft", "task_card"),
|
|
)
|
|
base.update(kw)
|
|
return CapabilityDescriptor(**base)
|
|
|
|
|
|
def _connected_adapter(**desc_kw):
|
|
desc = make_desc(**desc_kw)
|
|
stub = StubConnector(desc)
|
|
adapter = RelayAdapter(PlatformConfig(), desc, transport=stub)
|
|
return adapter, stub
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.get_event_loop().run_until_complete(coro)
|
|
|
|
|
|
@pytest.fixture()
|
|
def loop():
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Draft streaming op
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDraftOp:
|
|
def test_send_draft_emits_draft_op(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
result = loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=7, content="hel")
|
|
)
|
|
assert result.success
|
|
frames = [s for s in stub.sent if s.get("op") == "draft"]
|
|
assert len(frames) == 1
|
|
frame = frames[0]
|
|
assert frame["chat_id"] == "C1"
|
|
assert frame["draft_id"] == 7
|
|
assert frame["content"] == "hel"
|
|
assert frame["final"] is False
|
|
|
|
def test_send_draft_requires_descriptor_op(self, loop):
|
|
# supports_draft_streaming True but op NOT advertised: old connector.
|
|
adapter, stub = _connected_adapter(
|
|
supported_ops=("send", "edit", "typing")
|
|
)
|
|
with pytest.raises(NotImplementedError):
|
|
loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=1, content="x")
|
|
)
|
|
assert not [s for s in stub.sent if s.get("op") == "draft"]
|
|
|
|
def test_supports_draft_streaming_requires_both_flag_and_op(self):
|
|
both, _ = _connected_adapter()
|
|
assert both.supports_draft_streaming() is True
|
|
flag_only, _ = _connected_adapter(
|
|
supported_ops=("send", "edit", "typing")
|
|
)
|
|
assert flag_only.supports_draft_streaming() is False
|
|
# Legacy connector with EMPTY supported_ops: fail-open elsewhere, but
|
|
# draft must NOT fail open — the op did not exist pre-contract.
|
|
legacy, _ = _connected_adapter(supported_ops=())
|
|
assert legacy.supports_draft_streaming() is False
|
|
|
|
def test_final_send_seals_open_draft(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=7, content="hel")
|
|
)
|
|
loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=7, content="hello wor")
|
|
)
|
|
stub.next_draft_result = {"success": True, "message_id": "1723600000.1"}
|
|
result = loop.run_until_complete(
|
|
adapter.send("C1", "hello world", reply_to=None)
|
|
)
|
|
assert result.success
|
|
# No regular send op — the final frame rides the draft stream.
|
|
ops = [s["op"] for s in stub.sent]
|
|
assert "send" not in ops, "final must seal the stream, not post anew"
|
|
finals = [
|
|
s for s in stub.sent if s.get("op") == "draft" and s.get("final")
|
|
]
|
|
assert len(finals) == 1
|
|
assert finals[0]["content"] == "hello world"
|
|
# Stream ts comes back as the message identity.
|
|
assert result.message_id == "1723600000.1"
|
|
|
|
def test_send_without_open_draft_is_normal(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
loop.run_until_complete(adapter.send("C1", "plain", reply_to=None))
|
|
assert [s["op"] for s in stub.sent] == ["send"]
|
|
|
|
def test_draft_state_clears_after_seal(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=7, content="a")
|
|
)
|
|
loop.run_until_complete(adapter.send("C1", "a!", reply_to=None))
|
|
# Second send on the same chat is a NORMAL send again.
|
|
loop.run_until_complete(adapter.send("C1", "followup", reply_to=None))
|
|
ops = [s["op"] for s in stub.sent]
|
|
assert ops == ["draft", "draft", "send"]
|
|
|
|
def test_draft_failure_result_propagates(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
stub.next_draft_result = {"success": False, "error": "stream_gone"}
|
|
result = loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=1, content="x")
|
|
)
|
|
assert not result.success
|
|
# DEFINITE connector rejection disarms seal-interception: the
|
|
# stream consumer falls back to edit-based streaming on this
|
|
# failure, and its turn-final must go out as a REAL send — never
|
|
# a seal on a stream the connector just rejected.
|
|
stub.next_send_result = {"success": True, "message_id": "m2"}
|
|
loop.run_until_complete(adapter.send("C1", "final", reply_to=None))
|
|
assert [s["op"] for s in stub.sent][-1] == "send"
|
|
|
|
def test_draft_transport_exception_keeps_interception_armed(self, loop):
|
|
"""Ambiguity contract (G-D1): a transport EXCEPTION — as opposed
|
|
to an explicit rejection — may mean the frame was delivered, so
|
|
interception stays armed and the turn-final still seals."""
|
|
adapter, _ = _connected_adapter()
|
|
|
|
class _FlakyOnce:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
async def send_outbound(self, payload, platform=None):
|
|
self.calls += 1
|
|
if self.calls == 1:
|
|
raise ConnectionError("mid-write drop")
|
|
return {"success": True, "message_id": "ts.9"}
|
|
|
|
adapter._transport = _FlakyOnce()
|
|
result = loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=2, content="x")
|
|
)
|
|
assert not result.success
|
|
# Still armed: the turn-final converts to the sealing frame.
|
|
final = loop.run_until_complete(adapter.send("C1", "final", reply_to=None))
|
|
assert final.success
|
|
assert final.message_id == "ts.9"
|
|
|
|
def test_drafts_are_per_chat(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
loop.run_until_complete(
|
|
adapter.send_draft(chat_id="C1", draft_id=1, content="a")
|
|
)
|
|
# A send to a DIFFERENT chat is untouched.
|
|
loop.run_until_complete(adapter.send("D9", "other", reply_to=None))
|
|
by_op = [(s["op"], s["chat_id"]) for s in stub.sent]
|
|
assert ("send", "D9") in by_op
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task-card ops (#85476 TurnRunner seam, relay leg)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _tasks():
|
|
return [
|
|
{
|
|
"id": "call_1",
|
|
"title": "terminal",
|
|
"status": "in_progress",
|
|
"details": "ls -la",
|
|
},
|
|
{
|
|
"id": "call_2",
|
|
"title": "web_search",
|
|
"status": "complete",
|
|
"details": "slack startStream",
|
|
},
|
|
]
|
|
|
|
|
|
class TestTaskCardOps:
|
|
def test_progress_emits_task_card_op(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
result = loop.run_until_complete(
|
|
adapter.send_native_task_card_progress(
|
|
chat_id="C1", tasks=_tasks(), reply_to="1700.100"
|
|
)
|
|
)
|
|
assert result.success
|
|
frames = [s for s in stub.sent if s.get("op") == "task_card"]
|
|
assert len(frames) == 1
|
|
assert frames[0]["card_id"] == "turn:1700.100"
|
|
chunk_ids = [c["id"] for c in frames[0]["chunks"]]
|
|
assert chunk_ids == ["call_1", "call_2"]
|
|
statuses = {c["id"]: c["status"] for c in frames[0]["chunks"]}
|
|
assert statuses["call_2"] == "complete"
|
|
|
|
def test_stop_emits_task_card_stop(self, loop):
|
|
adapter, stub = _connected_adapter()
|
|
loop.run_until_complete(
|
|
adapter.send_native_task_card_progress(
|
|
chat_id="C1", tasks=_tasks(), reply_to="1700.100"
|
|
)
|
|
)
|
|
loop.run_until_complete(
|
|
adapter.stop_native_task_card_progress(chat_id="C1", reply_to="1700.100")
|
|
)
|
|
assert [s["op"] for s in stub.sent] == ["task_card", "task_card_stop"]
|
|
|
|
def test_task_card_gated_on_descriptor(self, loop):
|
|
adapter, stub = _connected_adapter(
|
|
supported_ops=("send", "edit", "typing", "draft")
|
|
)
|
|
# No "task_card" in supported_ops: the TurnRunner's hasattr gate must
|
|
# see NO capability — expose it via a probe method returning False,
|
|
# and the send must be a clean no-op failure (never an exception on
|
|
# the turn path).
|
|
assert adapter.supports_native_task_cards() is False
|
|
result = loop.run_until_complete(
|
|
adapter.send_native_task_card_progress(
|
|
chat_id="C1", tasks=_tasks(), reply_to="t"
|
|
)
|
|
)
|
|
assert not result.success
|
|
assert not stub.sent
|
|
|
|
def test_task_card_legacy_empty_ops_not_fail_open(self):
|
|
legacy, _ = _connected_adapter(supported_ops=())
|
|
assert legacy.supports_native_task_cards() is False
|