Fix interactive lifecycle and resume history

This commit is contained in:
0xallam
2026-04-26 12:26:48 -07:00
parent e8b172bd2a
commit 9f45121dce
4 changed files with 152 additions and 17 deletions
+5 -5
View File
@@ -236,11 +236,11 @@ def build_strix_agent(
instructions=instructions,
tools=tools,
tool_use_behavior=_finish_tool_use_behavior,
# Autonomous Strix runs must keep forcing tool calls after each
# tool result; otherwise the SDK resets tool_choice to auto and a
# plain-text model response can become final output before
# finish_scan / agent_finish.
reset_tool_choice=False,
# Non-interactive runs must keep forcing tool calls until the
# lifecycle tool completes. Interactive runs need the SDK default
# reset so a tool-assisted answer can end as plain text instead of
# looping through think/list_todos forever.
reset_tool_choice=interactive,
# model=None so ``RunConfig.model`` drives provider selection
# via :func:`build_multi_provider` rather than the SDK's default.
model=None,
+4 -1
View File
@@ -30,6 +30,9 @@ INTERACTIVE BEHAVIOR:
- EVERY message while working MUST contain exactly one tool call — this is what keeps execution moving. No tool call = execution stops.
- You may include brief explanatory text BEFORE the tool call
- Respond naturally when the user asks questions or gives instructions
- For simple conversation, acknowledgements, or direct questions that you can answer from current context, reply in plain text and stop. Do NOT call think just to prepare wording.
- If you use a tool to answer a user question (for example list_todos, view_agent_graph, or a file read), then after the tool result arrives, provide the answer in plain text and stop unless the user explicitly asked you to continue working.
- Never loop through think or other tools just to prepare, polish, confirm, or announce a final answer. Once you know the answer, say it.
- NEVER send empty messages — if you have nothing to do or say, call the wait_for_message tool
- If you catch yourself about to describe multiple steps without a tool call, STOP and call the think tool instead
{% else %}
@@ -149,7 +152,7 @@ OPERATIONAL PRINCIPLES:
- Use custom Python or shell code when you want to dig deeper, automate custom workflows, batch operations, triage results, build target-specific validation, or do work that existing tools do not cover cleanly
- Chain related weaknesses when needed to demonstrate real impact
- Consider business logic and context in validation
- NEVER skip think tool - it's your most important tool for reasoning and success
- Use think for non-trivial planning, uncertainty, multi-step security work, or choosing what to do next. Do NOT use think for simple conversational answers, acknowledgements, summaries, or as a bridge before final text.
- WORK METHODICALLY - Don't stop at shallow checks when deeper in-scope validation is warranted
- Continue iterating until the most promising in-scope vectors have been properly assessed
- Try multiple approaches simultaneously - don't wait for one to fail
+5 -1
View File
@@ -612,7 +612,11 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) ->
console.print("\n")
console.print(panel)
console.print()
console.print("[#60a5fa]strix.ai[/] [dim]·[/] [#60a5fa]discord.gg/strix-ai[/]")
console.print(
"[#60a5fa]strix.ai[/] [dim]·[/] "
"[#60a5fa]docs.strix.ai[/] [dim]·[/] "
"[#60a5fa]discord.gg/strix-ai[/]"
)
console.print()
+138 -10
View File
@@ -5,6 +5,7 @@ import contextlib
import json
import logging
import signal
import sqlite3
import sys
import threading
from collections.abc import Callable
@@ -674,6 +675,39 @@ class TuiLiveView:
parent_id=parent_of.get(agent_id) if isinstance(parent_of, dict) else None,
status=str(status),
)
self._hydrate_sdk_session_history(run_dir, statuses.keys())
def _hydrate_sdk_session_history(self, run_dir: Path, agent_ids: Any) -> None:
agents_db = run_dir / "agents.db"
session_ids = [aid for aid in agent_ids if isinstance(aid, str)]
if not agents_db.exists() or not session_ids:
return
session_id_set = set(session_ids)
try:
with sqlite3.connect(agents_db) as conn:
rows = conn.execute(
"select id, session_id, message_data, created_at "
"from agent_messages order by id"
).fetchall()
except sqlite3.Error:
logger.exception("Failed to hydrate TUI history from %s", agents_db)
return
for row_id, agent_id, message_data, created_at in rows:
if agent_id not in session_id_set:
continue
try:
item = json.loads(message_data)
except (TypeError, json.JSONDecodeError):
logger.debug("Skipping unreadable SDK session item %s for %s", row_id, agent_id)
continue
if not isinstance(item, dict):
continue
self._ingest_session_history_item(
str(agent_id),
item,
timestamp=_sqlite_timestamp_to_iso(created_at),
)
def upsert_agent(
self,
@@ -747,6 +781,53 @@ class TuiLiveView:
if delta:
self._record_assistant_message(agent_id, str(delta), final=False)
def _ingest_session_history_item(
self,
agent_id: str,
item: dict[str, Any],
*,
timestamp: str,
) -> None:
item_type = item.get("type")
role = item.get("role")
if role in {"user", "assistant"} and (item_type in {None, "message"}):
content = _session_message_text(item)
if content:
self._append_event(
agent_id,
"chat",
{
"role": role,
"content": content,
"metadata": {"source": "sdk_session"},
},
timestamp=timestamp,
)
return
if item_type == "function_call":
self._record_tool_call_data(
agent_id,
{
"call_id": str(item.get("call_id") or item.get("id") or ""),
"tool_name": str(item.get("name") or "tool"),
"args": _parse_json_object(item.get("arguments")),
},
timestamp=timestamp,
)
return
if item_type == "function_call_output":
self._record_tool_output_data(
agent_id,
{
"call_id": str(item.get("call_id") or item.get("id") or ""),
"tool_name": "tool",
"output": item.get("output"),
},
timestamp=timestamp,
)
def _record_assistant_message(self, agent_id: str, content: str, *, final: bool) -> None:
if not content:
return
@@ -775,7 +856,15 @@ class TuiLiveView:
self._bump_event(existing)
def _record_tool_call(self, agent_id: str, item: Any) -> None:
call = _sdk_tool_call_data(item)
self._record_tool_call_data(agent_id, _sdk_tool_call_data(item))
def _record_tool_call_data(
self,
agent_id: str,
call: dict[str, Any],
*,
timestamp: str | None = None,
) -> None:
call_id = call["call_id"]
existing = self._tool_event_by_call_id.get(call_id)
tool_data = {
@@ -786,14 +875,22 @@ class TuiLiveView:
"call_id": call_id,
}
if existing is None:
event = self._append_event(agent_id, "tool", tool_data)
event = self._append_event(agent_id, "tool", tool_data, timestamp=timestamp)
self._tool_event_by_call_id[call_id] = event
else:
existing["data"].update(tool_data)
self._bump_event(existing)
self._bump_event(existing, timestamp=timestamp)
def _record_tool_output(self, agent_id: str, item: Any) -> None:
output = _sdk_tool_output_data(item)
self._record_tool_output_data(agent_id, _sdk_tool_output_data(item))
def _record_tool_output_data(
self,
agent_id: str,
output: dict[str, Any],
*,
timestamp: str | None = None,
) -> None:
call_id = output["call_id"]
event = self._tool_event_by_call_id.get(call_id)
if event is None:
@@ -807,20 +904,28 @@ class TuiLiveView:
"agent_id": agent_id,
"call_id": call_id,
},
timestamp=timestamp,
)
self._tool_event_by_call_id[call_id] = event
result = _parse_json_value(output["output"])
event["data"]["result"] = result
event["data"]["status"] = _tool_status_from_result(result)
self._bump_event(event)
self._bump_event(event, timestamp=timestamp)
def _append_event(self, agent_id: str, event_type: str, data: dict[str, Any]) -> dict[str, Any]:
def _append_event(
self,
agent_id: str,
event_type: str,
data: dict[str, Any],
*,
timestamp: str | None = None,
) -> dict[str, Any]:
event = {
"id": f"{event_type}_{self._next_event_id}",
"type": event_type,
"agent_id": agent_id,
"timestamp": datetime.now(UTC).isoformat(),
"timestamp": timestamp or datetime.now(UTC).isoformat(),
"version": 0,
"data": data,
}
@@ -829,9 +934,9 @@ class TuiLiveView:
return event
@staticmethod
def _bump_event(event: dict[str, Any]) -> None:
def _bump_event(event: dict[str, Any], *, timestamp: str | None = None) -> None:
event["version"] = int(event.get("version", 0)) + 1
event["timestamp"] = datetime.now(UTC).isoformat()
event["timestamp"] = timestamp or datetime.now(UTC).isoformat()
def _sdk_tool_call_data(item: Any) -> dict[str, Any]:
@@ -859,10 +964,20 @@ def _sdk_tool_output_data(item: Any) -> dict[str, Any]:
def _sdk_message_text(item: Any) -> str:
raw = getattr(item, "raw_item", None)
content = _raw_field(raw, "content", [])
return _message_content_text(_raw_field(raw, "content", []))
def _session_message_text(item: dict[str, Any]) -> str:
return _message_content_text(item.get("content", ""))
def _message_content_text(content: Any) -> str:
parts: list[str] = []
content_items = content if isinstance(content, list) else [content]
for part in content_items:
if isinstance(part, str):
parts.append(part)
continue
text = _raw_field(part, "text")
if isinstance(text, str):
parts.append(text)
@@ -895,6 +1010,19 @@ def _tool_status_from_result(result: Any) -> str:
return "completed"
def _sqlite_timestamp_to_iso(value: Any) -> str:
if not isinstance(value, str) or not value.strip():
return datetime.now(UTC).isoformat()
text = value.strip()
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return text
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=UTC)
return parsed.astimezone(UTC).isoformat()
class QuitScreen(ModalScreen): # type: ignore[misc]
def compose(self) -> ComposeResult:
yield Grid(