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
4 changes: 2 additions & 2 deletions td_component/.tox-api-source-hash.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"tox_source_hash": "920c04f27ef8e11fd690fed427bba8971ac141a0629799e0fb30dc77589fe4d9",
"built_at": "2026-05-11T14:14:37.138801+00:00",
"tox_source_hash": "10ba287545a6f421b9deef39f586ad81ab51ad5af95d5474967ef2a2f8316874",
"built_at": "2026-05-11T14:33:59.514280+00:00",
"source_files": [
"td_component/tdpilot_api_agent.py",
"td_component/tdpilot_api_dispatcher.py",
Expand Down
Binary file modified td_component/tdpilot_API.tox
Binary file not shown.
78 changes: 50 additions & 28 deletions td_component/tdpilot_api_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -599,34 +599,11 @@ def _loop(self) -> str | None:
except Exception as exc: # noqa: BLE001
print(f"[tdpilot_API/agent] rollback_guard.__exit__ raised: {exc}")

# Phase 1.1 — if the guard fired a rollback, append a hint to
# the LAST tool_result so the LLM sees the regression context
# on its next API call. The hint goes in the tool_result's
# content (which can be a list of blocks) — keeps the
# alternating user/assistant constraint intact and pairs the
# hint with the failing batch's results.
if (
rollback_guard is not None
and getattr(rollback_guard, "rollback_fired", False)
and getattr(rollback_guard, "hint_text", "")
and results_block
):
hint = rollback_guard.hint_text
last = results_block[-1]
existing = last.get("content")
if isinstance(existing, str):
last["content"] = existing + "\n\n" + hint
elif isinstance(existing, list):
last["content"] = list(existing) + [{"type": "text", "text": hint}]
else:
last["content"] = hint
# Surface to the chat UI too — same callback the agent's
# natural-language text uses, so the user sees a yellow
# inline notice in the assistant bubble.
try:
self.on_text(hint)
except Exception: # noqa: BLE001
pass
# Phase 1.1 — append any rollback hint emitted by the guard
# onto the last tool_result + surface it through on_text.
# Logic extracted into ``_apply_rollback_hint`` for direct
# unit-testing (Codex P2 followup on PR #34).
self._apply_rollback_hint(rollback_guard, results_block)

self.messages.append({"role": "user", "content": results_block})

Expand All @@ -637,6 +614,51 @@ def _loop(self) -> str | None:

raise TurnBudgetExceeded(f"Tool-use loop exceeded turn_budget={self.turn_budget}")

# ------------------------------------------------------------------
# Phase 1.1 — auto-rollback hint plumbing
# ------------------------------------------------------------------

def _apply_rollback_hint(self, rollback_guard: Any, results_block: list[dict]) -> None:
"""Append a rollback hint to the last tool_result + surface via
``on_text``. No-op if ``rollback_guard`` is None, has no
``hint_text``, or ``results_block`` is empty.

Codex P2 review on PR #34 (2026-05-11) flagged that the prior
in-line condition keyed on ``rollback_fired`` — which is False
in the degraded path where the guard detected a regression but
couldn't actually open / undo the block. The bug dropped the
only signal the LLM would receive about that failure mode,
leaving it to continue from a broken graph state with no
feedback. Keying on ``hint_text`` (which the guard populates in
BOTH the success and the degraded paths) surfaces both.

Insertion strategy: the hint text-block gets appended to the
LAST tool_result's content (which may be a string or a list of
blocks). This preserves Anthropic's alternating user/assistant
constraint AND pairs the hint with the failing batch's
results — exactly where the LLM is most likely to attend on
its next API call. Also surfaced to the chat UI via the
``on_text`` callback so the user sees a yellow inline notice
in the assistant bubble.
"""
if rollback_guard is None:
return
hint = getattr(rollback_guard, "hint_text", "")
if not hint or not results_block:
return
last = results_block[-1]
existing = last.get("content")
if isinstance(existing, str):
last["content"] = existing + "\n\n" + hint
elif isinstance(existing, list):
last["content"] = list(existing) + [{"type": "text", "text": hint}]
else:
last["content"] = hint
try:
self.on_text(hint)
except Exception: # noqa: BLE001 — chat-side callback must never break the agent loop
pass

# ------------------------------------------------------------------
# Dynamic context (Phase 0.1)
# ------------------------------------------------------------------
Expand Down
84 changes: 71 additions & 13 deletions td_component/tdpilot_api_rollback.py
Original file line number Diff line number Diff line change
Expand Up @@ -358,13 +358,43 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
self.new_critical_count = int(diff.get("count", 0))

if self.new_critical_count > 0 and self._undo_block_opened:
# Roll back the entire batch atomically.
# Attempt the rollback and **inspect the dispatcher's return
# value** before claiming success. Codex P1 review on PR #34
# (2026-05-11): the prior version set rollback_fired = True
# unconditionally when undo was attempted, so if ui.undo.undo()
# raised or returned an error payload the user + LLM were
# both told "reverted" while the network stayed broken — a
# silent-correctness bug, not a crash.
#
# ``auto_rollback_end`` returns one of:
# * {"ok": True, "rolled_back": True} -> revert succeeded
# * {"ok": True, "rolled_back": False} -> we asked not to undo
# * {"ok": False, ...} -> undo() raised
# * {"error": "..."} -> endBlock() raised
# …or raises (network drop, dispatcher bug). Only the first
# shape counts as a successful rollback.
end_result: Any = None
try:
self._dispatcher("auto_rollback_end", {"undo": True})
except Exception: # noqa: BLE001
pass
self.rollback_fired = True
self.hint_text = format_hint(diff)
end_result = self._dispatcher("auto_rollback_end", {"undo": True})
except Exception as exc: # noqa: BLE001
end_result = {"error": f"{type(exc).__name__}: {exc}"}
if (
isinstance(end_result, dict)
and end_result.get("ok") is True
and end_result.get("rolled_back") is True
):
self.rollback_fired = True
self.hint_text = format_hint(diff, rolled_back=True)
else:
# Rollback was attempted but didn't succeed (handler returned
# error/non-success, or raised). Network is still broken;
# tell the truth in both signals.
self.rollback_fired = False
self.hint_text = format_hint(
diff,
rolled_back=False,
end_failure=_format_end_failure(end_result),
)
elif self.new_critical_count > 0:
# Detected regression but couldn't open the undo block —
# surface a hint without claiming rollback happened.
Expand All @@ -380,22 +410,50 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
return False # never swallow exceptions from the with-block


def format_hint(diff: dict, rolled_back: bool = True) -> str:
def _format_end_failure(end_result: Any) -> str:
"""Render a short ``(reason: ...)`` clause for the
rollback-could-not-apply branch of ``format_hint``. Used by
``AutoRollbackGuard.__exit__`` when ``auto_rollback_end`` came
back without an ``ok: True, rolled_back: True`` payload (Codex
P1 finding on PR #34: previously we lied about success in this
case)."""
if not isinstance(end_result, dict):
return ""
if end_result.get("undo_error"):
return f" (undo raised: {end_result['undo_error']})"
if end_result.get("error"):
return f" (endBlock raised: {end_result['error']})"
return ""


def format_hint(
diff: dict,
rolled_back: bool = True,
end_failure: str = "",
) -> str:
"""Render the hint message appended to the last tool_result on a
rollback fire. Intentionally short — costs DeepSeek output tokens
on the next turn (the model has to read + reason about it)."""
on the next turn (the model has to read + reason about it).

``end_failure`` is an optional short clause appended to the
"could not be applied" message describing what specifically
failed (Codex P1 followup — surface enough detail that the LLM
can decide whether to retry, abort, or escalate).
"""
n = int(diff.get("count", 0))
items = list(diff.get("new_criticals", []))
if not items:
verb = "reverted" if rolled_back else "detected"
return f"[tdpilot_auto_rollback] {n} new critical errors {verb}."
names = ", ".join(f"{it.get('path')} ({it.get('error_preview')})" for it in items[:3])
more = "" if len(items) <= 3 else f", +{len(items) - 3} more"
action = (
"the changes were automatically reverted via TD's undo"
if rolled_back
else "the rollback could not be applied (undo block not open) so the errors remain"
)
if rolled_back:
action = "the changes were automatically reverted via TD's undo"
else:
# The "(reason: ...)" detail is what distinguishes "we never
# opened the block" from "we opened it but undo() failed".
# The LLM can use this to pick a recovery strategy.
action = "the rollback could not be applied so the errors remain" + end_failure
return (
f"[tdpilot_auto_rollback] This batch introduced {n} new critical "
f"error(s) so {action}. Errors: {names}{more}. "
Expand Down
Loading
Loading