Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 63 additions & 5 deletions bot/telegram/bloc_bridge/bridge.py
Original file line number Diff line number Diff line change
Expand Up @@ -471,15 +471,17 @@ def _terminal_request_summary(request, state, response=None, error=None):
else:
tool_names = getattr(error, "tool_names", None)
tool_call_count = getattr(error, "tool_call_count", "unknown")
tool_count = len(tool_names) if tool_names is not None else "unknown"
tool_summary = ", ".join(tool_names) if tool_names is not None and tool_names else "none"
if tool_names is None:
tool_summary = "unknown"
elapsed = time.time() - request.get("created_at", time.time())
return "\n".join(
(
f"Request {request['reference']}",
f"Status: {_request_status_text(state)}",
f"Ref: {request['reference']}",
cost_label,
f"Tools used: {tool_summary}; {tool_call_count} calls",
f"Status: {_request_status_text(state)}",
f"Duration: {_format_elapsed(elapsed)}",
f"Tools used: {tool_count} distinct; {tool_call_count} calls",
)
)

Expand Down Expand Up @@ -968,7 +970,6 @@ def progress(text):
else:
set_phase("opencode_request")
response = answer_async(chat_id, prompt, sessions, reference, set_phase, progress)
response = f"{response.rstrip()}\n\nRef: {reference}"
request_store.update(reference, state="delivering", detail="completed")
set_phase("telegram_final_send")
send(chat_id, response)
Expand Down Expand Up @@ -1108,6 +1109,58 @@ def _run_with_slot(function, *args):
_REQUEST_SLOTS.release()


def _command_worker(chat_id, prompt, sessions, update_id):
"""Run a non-model slash command without creating a request card."""
try:
bridge_runtime.check_shutdown()
response = authorized_response(chat_id, prompt, sessions)
bridge_runtime.check_shutdown()
send(chat_id, response)
except bridge_runtime.ShutdownRequested:
return
except Exception as error:
print(
f"command_failed update_id={update_id} error_type={type(error).__name__}",
file=sys.stderr,
flush=True,
)
try:
send(chat_id, "The command failed. Please try again later.")
except Exception as delivery_error:
print(
f"command_reply_failed update_id={update_id} error_type={type(delivery_error).__name__}",
file=sys.stderr,
flush=True,
)


def _enqueue_command(chat_id, prompt, sessions, update_id):
"""Queue a non-model command reply without acknowledging it with a card."""
if not _REQUEST_SLOTS.acquire(blocking=False):
send(chat_id, "The bridge is busy. Please retry later.")
return
try:
_REQUEST_EXECUTOR.submit(
_run_with_slot,
_command_worker,
chat_id,
prompt,
sessions,
update_id,
)
except bridge_runtime.ShutdownRequested:
_REQUEST_SLOTS.release()
raise
except Exception as error:
_REQUEST_SLOTS.release()
print(
f"command_enqueue_failed update_id={update_id} error_type={type(error).__name__}",
file=sys.stderr,
flush=True,
)
send(chat_id, "The command could not be queued. Please try again later.")


def _recovery_notice_worker(chat_id, update_id, reference, acknowledgement_id):
try:
edit_reply(
Expand Down Expand Up @@ -1327,6 +1380,11 @@ def handle_update(update, sessions):
)
send(chat_id, "The request control command failed. Use /status to check the request.")
return
if command is not None:
if not request_store.claim_update(update_id):
return
_enqueue_command(chat_id, prompt, sessions, update_id)
return
_acknowledge_and_enqueue(chat_id, prompt, sessions, update_id)


Expand Down
5 changes: 1 addition & 4 deletions bot/telegram/bloc_bridge/opencode/cost_reporting.py
Original file line number Diff line number Diff line change
Expand Up @@ -223,10 +223,7 @@ def format_exchange_footer(messages):
if usage["message_count"] == 0 or usage["cost"] is None:
return "\n\nCost: unavailable"
details = usage["agents"]
if len(details) == 1:
agent, cost = next(iter(details.items()))
return f"\n\nCost: {format_cost(usage['cost'])} ({agent}: {format_cost(cost)})"
if details:
if len(details) > 1:
breakdown = ", ".join(f"{agent}: {format_cost(cost)}" for agent, cost in sorted(details.items()))
return f"\n\nCost: {format_cost(usage['cost'])} ({breakdown})"
return f"\n\nCost: {format_cost(usage['cost'])}"
Expand Down
17 changes: 11 additions & 6 deletions bot/telegram/bloc_bridge/opencode/exchange.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@

_SESSIONS_LOCK = threading.RLock()
_PROGRESS_EVENT_MIN_INTERVAL_SECONDS = 5
_PROGRESS_AGE_DISPLAY_THRESHOLD_SECONDS = 5
_PERMISSION_POLL_INTERVAL_SECONDS = 5
_PERMISSION_SESSION_LOOKUP_TIMEOUT_SECONDS = 3
_PERMISSION_ANCESTRY_MAX_DEPTH = 8
Expand Down Expand Up @@ -710,12 +711,16 @@ def _report_progress(self, snapshot, progress, api_responding=True):
text = f"OpenCode is waiting for {count} {permission_label} ({elapsed}s elapsed)."
else:
text = f"{self.current_stage or 'OpenCode is working on your request'} ({elapsed}s elapsed)."
api_heartbeat = (
f"API heartbeat: {api_age}s ago"
if api_responding
else f"API last responded: {api_age}s ago"
)
text = f"{text}\nOpenCode activity: {activity_age}s ago; {api_heartbeat}."
if (
activity_age > _PROGRESS_AGE_DISPLAY_THRESHOLD_SECONDS
or api_age > _PROGRESS_AGE_DISPLAY_THRESHOLD_SECONDS
):
api_heartbeat = (
f"API heartbeat: {api_age}s ago"
if api_responding
else f"API last responded: {api_age}s ago"
)
text = f"{text}\nOpenCode activity: {activity_age}s ago; {api_heartbeat}."
cost_line = observed_cost or self.reported_cost
if cost_line:
text = f"{text}\n{cost_line}"
Expand Down
14 changes: 10 additions & 4 deletions bot/telegram/bloc_bridge/telegram/request_presentation.py
Original file line number Diff line number Diff line change
Expand Up @@ -177,11 +177,17 @@ def _next_available_letter(self, used_letters):

def _render_active(self):
self.expire_statuses()
lines = [
self.progress,
f"Ref: {self.reference}",
"Use /status or /abort (/cancel); buttons below control this request.",
progress_lines = self.progress.splitlines()
cost_lines = [
line for line in progress_lines if line.startswith(("Cost observed so far:", "Final cost:", "Cost:"))
]
tool_lines = [line for line in progress_lines if line.startswith("Tools used:")]
action_lines = [
line
for line in progress_lines
if line not in cost_lines and line not in tool_lines
]
lines = [f"Ref: {self.reference}", *cost_lines, *tool_lines, *action_lines]
if self.status_entries:
lines.extend(("", "Status checks:", *(entry["text"] for entry in self.status_entries)))
if self.permissions:
Expand Down
37 changes: 28 additions & 9 deletions tests/test_telegram_bridge.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,26 @@ def test_permission_letters_rotate_after_a_request_is_resolved(self):

self.assertEqual(presentation.permissions_by_letter(), {"B": second})

def test_active_card_orders_reference_cost_tools_and_action_without_command_hints(self):
presentation = RequestPresentation(123, 456, "REF12345")
presentation.set_progress(
"Working on the request.\n"
"OpenCode activity: 2s ago; API heartbeat: 1s ago.\n"
"Cost observed so far: $0.125000 (partial)\n"
"Tools used: read, bash"
)

text, keyboard = presentation.edit_payload()
lines = text.splitlines()

self.assertEqual(lines[0], "Ref: REF12345")
self.assertEqual(lines[1], "Cost observed so far: $0.125000 (partial)")
self.assertEqual(lines[2], "Tools used: read, bash")
self.assertEqual(lines[3], "Working on the request.")
self.assertNotIn("/cancel", text)
self.assertNotIn("/abort", text)
self.assertEqual([button["text"] for button in keyboard["inline_keyboard"][0]], ["Status", "Abort"])

def test_http_requests_do_not_add_basic_authentication(self):
for url in (bridge_config.OC + "/global/health", bridge_config.TG + "getMe"):
with self.subTest(url=url):
Expand Down Expand Up @@ -223,7 +243,7 @@ def test_answer_creates_and_persists_a_session(self):
with patch.object(session_state, "save_sessions") as save:
result = bridge.answer(123, "Hi", {})

self.assertEqual(result, "Hello\n\nCost: $0.100000 (build: $0.100000)")
self.assertEqual(result, "Hello\n\nCost: $0.100000")
title.assert_called_once_with()
request.assert_has_calls(
[
Expand Down Expand Up @@ -285,8 +305,8 @@ def test_handle_update_sends_ack_before_enqueuing_worker(self):

send.assert_called_once()
self.assertIn("Ref:", send.call_args.args[1])
self.assertIn("/status", send.call_args.args[1])
self.assertIn("/abort", send.call_args.args[1])
self.assertNotIn("/status", send.call_args.args[1])
self.assertNotIn("/abort", send.call_args.args[1])
keyboard = send.call_args.kwargs["reply_markup"]["inline_keyboard"]
self.assertEqual([button["text"] for button in keyboard[0]], ["Status", "Abort"])
submit.assert_called_once()
Expand Down Expand Up @@ -476,8 +496,7 @@ def event_stream():
self.assertEqual(
progress,
[
"Analysis in progress (0s elapsed).\n"
"OpenCode activity: 0s ago; API heartbeat: 0s ago."
"Analysis in progress (0s elapsed)."
],
)
self.assertNotIn("private reasoning content", " ".join(progress))
Expand Down Expand Up @@ -926,15 +945,15 @@ def test_cost_starts_at_zero_after_creating_a_new_session(self):
self.assertIn("$0.000000", report)
history.assert_called_once_with("new-session")

def test_worker_does_not_forward_unknown_slash_commands_to_the_llm(self):
def test_command_worker_does_not_forward_unknown_slash_commands_to_the_llm(self):
with patch.object(bridge, "authorized_response", return_value="Use /help") as respond:
with patch.object(bridge, "answer_async") as answer:
with patch.object(bridge, "edit_reply"):
with patch.object(bridge, "send"):
bridge._request_worker(123, "/unknown", {}, 991, "REF12345", 657)
with patch.object(bridge, "send") as send:
bridge._command_worker(123, "/unknown", {}, 991)

respond.assert_called_once_with(123, "/unknown", {})
answer.assert_not_called()
send.assert_called_once_with(123, "Use /help")

def test_health_command_reports_only_sanitized_status_without_a_model_call(self):
with patch.object(http_api, "http", return_value={"healthy": True}) as request:
Expand Down
34 changes: 32 additions & 2 deletions tests/test_telegram_health_feedback.py
Original file line number Diff line number Diff line change
Expand Up @@ -328,6 +328,36 @@ def test_progress_distinguishes_stale_agent_activity_from_live_api_heartbeat(sel
self.assertIn("OpenCode activity: 30s ago; API heartbeat: 0s ago.", self.progress[0][1])
self.assertIn("OpenCode activity: 60s ago; API heartbeat: 0s ago.", self.progress[1][1])

def test_progress_hides_recent_api_and_opencode_ages_until_over_five_seconds(self):
snapshot = bridge.exchange._MonitorSnapshot(True, False, [], [])
updates = []

for age in (4, 5, 6):
monitor = bridge._AsyncRequestMonitor(
"private-session",
"private-prompt",
"REF12345",
time.time(),
)
monitor.current_stage = "Analysis in progress"
monitor.monitor_started_at = 0
monitor.last_progress_at = -30
monitor.last_opencode_activity_at = 0
monitor.last_api_success_at = 0
with patch.object(bridge.time, "monotonic", return_value=age):
monitor._report_progress(snapshot, updates.append)

self.assertEqual(updates[0], "Analysis in progress (4s elapsed).")
self.assertEqual(
updates[1],
"Analysis in progress (5s elapsed).",
)
self.assertEqual(
updates[2],
"Analysis in progress (6s elapsed).\n"
"OpenCode activity: 6s ago; API heartbeat: 6s ago.",
)

def test_progress_reports_age_of_last_api_response_during_polling_errors(self):
prompt = {"info": {"id": "msg_prompt", "role": "user"}}

Expand Down Expand Up @@ -457,8 +487,8 @@ def exchange(_chat, _prompt, _sessions, _ref, _phase, progress):
patch("sys.stderr", output):
bridge._request_worker(123, "Hi", {}, 123, "REF12345", 456)
self.assertEqual(edit.call_count, 2)
self.assertIn("/status", edit.call_args_list[0].args[2])
self.assertIn("/abort", edit.call_args_list[0].args[2])
self.assertNotIn("/status", edit.call_args_list[0].args[2])
self.assertNotIn("/abort", edit.call_args_list[0].args[2])
self.assertIn("REF12345", edit.call_args_list[0].args[2])
self.assertIn("timed out", send.call_args.args[1])
self.assertIn("phase=response_wait", output.getvalue())
Expand Down
29 changes: 24 additions & 5 deletions tests/test_telegram_requests.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,24 @@ def test_duplicate_telegram_update_does_not_acknowledge_or_enqueue_twice(self):
send.assert_called_once()
submit.assert_called_once()

def test_non_model_command_sends_only_its_result_without_a_request_card(self):
update = {"update_id": 1031, "message": {"chat": {"id": 123}, "text": "/start"}}
with patch.object(bridge, "send") as send:
with patch.object(bridge._REQUEST_EXECUTOR, "submit") as submit:
with patch.object(bridge, "_REQUEST_SLOTS") as slots:
slots.acquire.return_value = True
bridge.handle_update(update, {})
worker, *args = submit.call_args.args
worker(*args)

self.assertEqual(submit.call_args.args[0], bridge._run_with_slot)
self.assertEqual(submit.call_args.args[1], bridge._command_worker)
self.assertEqual(send.call_count, 1)
self.assertIn("Welcome!", send.call_args.args[1])
self.assertNotIn("Ref:", send.call_args.args[1])
self.assertNotIn("reply_markup", send.call_args.kwargs)
self.assertIsNone(request_store.get_by_update_id(1031))

def test_status_is_private_to_the_chat_and_does_not_expose_session_ids(self):
request, _ = request_store.create("C" * 16, 123, 104, state="running")
request_store.update(
Expand Down Expand Up @@ -114,7 +132,8 @@ def test_status_command_appends_timestamped_update_to_request_card(self):
self.assertEqual(edit.call_count, 1)
self.assertIn("Status checks:", edit.call_args.args[2])
self.assertRegex(edit.call_args.args[2], r"\[\d\d:\d\d:\d\d UTC\] Request .+: running in OpenCode\.")
self.assertIn("/status or /abort", edit.call_args.args[2])
self.assertNotIn("/cancel", edit.call_args.args[2])
self.assertNotIn("/abort", edit.call_args.args[2])
self.assertEqual(edit.call_args.kwargs["reply_markup"]["inline_keyboard"][0][0]["text"], "Status")
bridge._forget_request_presentation(request["reference"])

Expand Down Expand Up @@ -503,7 +522,7 @@ def test_recovered_prompt_returns_completed_response_without_new_submission(self
self.assertIn("Final cost: unavailable", edit.call_args.args[2])
self.assertEqual(
send.call_args.args[1],
"Recovered result\n\nCost: unavailable\n\nTools used: none\nTool calls: 0\n\nRef: 2222222222222222",
"Recovered result\n\nCost: unavailable\n\nTools used: none\nTool calls: 0",
)
self.assertEqual(request_store.get(request["reference"])["state"], "completed")
submit.assert_not_called()
Expand All @@ -525,9 +544,9 @@ def test_final_cost_reply_keeps_reference_after_acknowledgement_edit(self):

self.assertIn("Final cost: $0.250000", edit.call_args.args[2])
self.assertIn(request["reference"], edit.call_args.args[2])
self.assertIn("Tools used: 0 distinct; 0 calls", edit.call_args.args[2])
self.assertIn("Tools used: none; 0 calls", edit.call_args.args[2])
self.assertEqual(edit.call_args.kwargs["reply_markup"], {"inline_keyboard": []})
self.assertEqual(send.call_args.args[1], f"Done\n\nCost: $0.250000\n\nRef: {request['reference']}")
self.assertEqual(send.call_args.args[1], "Done\n\nCost: $0.250000")

def test_recovery_does_not_replay_an_unconfirmed_submission(self):
request, _ = request_store.create("4" * 16, 123, 111, state="running")
Expand Down Expand Up @@ -658,7 +677,7 @@ def test_monitor_reports_timeout_when_open_code_records_abort_as_assistant_error
self.assertEqual(request_store.get(request["reference"])["state"], "timed_out")
self.assertIn("timed out and stopped", bridge.request_status_response(123, request["reference"]))
self.assertIn("Status: timed out and stopped", edit.call_args.args[2])
self.assertIn("Tools used: 2 distinct; 1 calls", edit.call_args.args[2])
self.assertIn("Tools used: bash, gh; 1 calls", edit.call_args.args[2])
self.assertIn("reached the configured 5-day limit", send.call_args.args[1])
self.assertIn("Tools used: bash, gh", send.call_args.args[1])

Expand Down
14 changes: 5 additions & 9 deletions tests/test_telegram_shutdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,25 +75,22 @@ def request(url, payload=None, timeout=None):
updates = [
{
"update_id": i,
"message": {"chat": {"id": 123}, "text": "/xx" if scenario == "command" else "Hello"},
"message": {"chat": {"id": 123}, "text": "/start" if scenario == "command" else "Hello"},
}
for i in range(count)
]
return {"ok": True, "result": updates}
if scenario in ("worker", "queued", "command"):
if scenario == "command":
bridge._REQUEST_EXECUTOR._tasks.join()
if not active.wait(3):
raise AssertionError("Worker did not start")
return block()
return blocked.wait(60)
if url == TG + "sendMessage":
if scenario == "ack":
return block()
if scenario == "command":
active.set()
return {"ok": True, "result": {"message_id": 456}}
if url == TG + "editMessageText" and scenario == "command":
active.set()
return {"ok": True, "result": True}
raise AssertionError("Unexpected HTTP call")

def status(timeout):
Expand Down Expand Up @@ -153,10 +150,9 @@ def assert_clean_shutdown(self, scenario):
self.assertEqual(process.returncode, 0, stderr)
self.assertIn("Received SIGTERM", stderr)
self.assertIn("STOPPED", stdout)
self.assertNotIn("request_completed", stderr)
if scenario == "command":
self.assertIn("request_completed", stderr)
else:
self.assertNotIn("request_completed", stderr)
self.assertNotIn("request_accepted", stderr)
if scenario == "queued":
self.assertEqual(stderr.count("request_accepted"), 1)
finally:
Expand Down
Loading