diff --git a/agent-ops/.version b/agent-ops/.version index b131245..7df4129 100644 --- a/agent-ops/.version +++ b/agent-ops/.version @@ -1 +1 @@ -1.1.203 +1.1.204 diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json b/agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json index faec9cf..a882f85 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json +++ b/agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json @@ -93,6 +93,9 @@ ], "output_format": "jsonl", "session_stall_resume": true, + "auxiliary_logs": [ + "/app/opencode-data/opencode/log/opencode.log" + ], "environment": { "TMPDIR": "/tmp" } @@ -141,6 +144,9 @@ ], "output_format": "jsonl", "session_stall_resume": true, + "auxiliary_logs": [ + "/app/opencode-data/opencode/log/opencode.log" + ], "environment": { "TMPDIR": "/tmp" } @@ -189,6 +195,9 @@ ], "output_format": "jsonl", "session_stall_resume": true, + "auxiliary_logs": [ + "/app/opencode-data/opencode/log/opencode.log" + ], "environment": { "TMPDIR": "/tmp" } diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py b/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py index 11e06fa..90f7284 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py +++ b/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py @@ -2713,6 +2713,35 @@ def json_agent_terminal_outcome_from_line( return None +def json_agent_failure_diagnostic(value: object) -> str | None: + """Return only the terminal assistant failure, never prior conversation text.""" + if not isinstance(value, dict) or str(value.get("type", "")).lower() != "agent_end": + return None + if value.get("willRetry") is True: + return None + messages = value.get("messages") + if not isinstance(messages, list): + return None + for message in reversed(messages): + if ( + not isinstance(message, dict) + or str(message.get("role", "")).lower() != "assistant" + ): + continue + stop_reason = message.get("stopReason") or message.get("stop_reason") + if str(stop_reason or "").lower() not in {"error", "aborted"}: + return None + diagnostic = { + "type": "agent_end", + "stopReason": stop_reason, + } + for field in ("errorMessage", "error_message", "error", "code"): + if field in message: + diagnostic[field] = message[field] + return json.dumps(diagnostic, ensure_ascii=False) + return None + + def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None: if channel == "stderr": return line @@ -2726,9 +2755,10 @@ def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None: severity = str(value.get("severity") or value.get("level") or "").lower() status = str(value.get("status") or "").lower() subtype = str(value.get("subtype") or "").lower() + if event_type == "agent_end": + return json_agent_failure_diagnostic(value) if ( - json_agent_terminal_outcome(value) == "failed" - or event_type in {"error", "fatal", "request.failed", "turn.failed", "rate_limit_event"} + event_type in {"error", "fatal", "request.failed", "turn.failed", "rate_limit_event"} or severity in {"error", "fatal"} or subtype.startswith("error") or bool(value.get("is_error")) @@ -2810,11 +2840,19 @@ async def terminate_process_group( pass -def auxiliary_log_diagnostics(path: Path) -> list[str]: +def auxiliary_log_diagnostics(path: Path, start_offset: int = 0) -> list[str]: if not path.exists(): return [] + try: + size = path.stat().st_size + offset = start_offset if 0 <= start_offset <= size else 0 + with path.open("rb") as handle: + handle.seek(offset) + text = handle.read().decode("utf-8", errors="replace") + except OSError: + return [] diagnostics: list[str] = [] - for line in path.read_text(encoding="utf-8", errors="replace").splitlines()[-200:]: + for line in text.splitlines()[-200:]: failure_class, evidence = classify_failure_with_evidence(line) if ( failure_class not in RECOVERABLE_RUNTIME_FAILURES @@ -2828,6 +2866,7 @@ def auxiliary_log_diagnostics(path: Path) -> list[str]: r"|\btoo many requests\b" r"|(?:rate.?limit|quota|capacity).{0,40}" r"(?:exceed|exhaust|reached|reject)" + r"|usage limit.{0,40}(?:exceed|exhaust|reached|reject)" r"|(?:exceed|exhaust|reached|reject).{0,40}" r"(?:rate.?limit|quota|capacity)" r"|(?:rate.?limit|quota).{0,40}retry after" @@ -2863,11 +2902,14 @@ def attempt_terminal_diagnostics( diagnostic = terminal_diagnostic(spec.cli, channel, payload) if diagnostic: diagnostics.append((f"{spec.cli}:{channel}", diagnostic)) + offsets = record.get("auxiliary_log_offsets", {}) for raw_path in record.get("auxiliary_logs", []): path = Path(str(raw_path)) diagnostics.extend( (f"{spec.cli}:auxiliary-log", diagnostic) - for diagnostic in auxiliary_log_diagnostics(path) + for diagnostic in auxiliary_log_diagnostics( + path, int(offsets.get(str(path), 0)) + ) ) return diagnostics @@ -3568,6 +3610,33 @@ async def invoke( ) started_at = now_iso() work_log_path = milestone_work_log_path(task) + auxiliary_logs = [ + str(item).format_map( + { + "agent": spec.cli, + "attempt_dir": str(attempt_dir), + "model": spec.model, + "prompt": "", + "reasoning_effort": str(spec.reasoning_effort or ""), + "resume_session": str(effective_resume_session or ""), + "resume_session_dir": ( + str(effective_resume_session_dir) + if effective_resume_session_dir is not None + else "" + ), + "session_id": session_id, + "target_id": str(spec.target_id or ""), + "workspace": str(workspace), + } + ) + for item in spec.runtime.get("auxiliary_logs", []) + ] + auxiliary_log_offsets = {} + for raw_path in auxiliary_logs: + try: + auxiliary_log_offsets[raw_path] = Path(raw_path).stat().st_size + except OSError: + auxiliary_log_offsets[raw_path] = 0 record: dict[str, Any] = { "execution_id": identity, "task": task.name, @@ -3601,27 +3670,8 @@ async def invoke( "stream_log": str(stream_path), "normalized_output_log": str(normalized_output_path), "heartbeat_log": str(heartbeat_path), - "auxiliary_logs": [ - str(item).format_map( - { - "agent": spec.cli, - "attempt_dir": str(attempt_dir), - "model": spec.model, - "prompt": "", - "reasoning_effort": str(spec.reasoning_effort or ""), - "resume_session": str(effective_resume_session or ""), - "resume_session_dir": ( - str(effective_resume_session_dir) - if effective_resume_session_dir is not None - else "" - ), - "session_id": session_id, - "target_id": str(spec.target_id or ""), - "workspace": str(workspace), - } - ) - for item in spec.runtime.get("auxiliary_logs", []) - ], + "auxiliary_logs": auxiliary_logs, + "auxiliary_log_offsets": auxiliary_log_offsets, "work_log": str(work_log_path.resolve()), "started_at": started_at, "status": "running", @@ -4076,7 +4126,10 @@ async def invoke( raise for raw_path in record.get("auxiliary_logs", []): - aux_diagnostics = auxiliary_log_diagnostics(Path(str(raw_path))) + aux_diagnostics = auxiliary_log_diagnostics( + Path(str(raw_path)), + int(record.get("auxiliary_log_offsets", {}).get(str(raw_path), 0)), + ) diagnostics.extend(aux_diagnostics) diagnostic_origins.extend( f"{spec.cli}:auxiliary-log" for _ in aux_diagnostics diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py b/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py index e675140..aee86e7 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py +++ b/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py @@ -394,6 +394,26 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase): self.assertEqual(failure, "provider-quota") self.assertIsNotNone(evidence) + def test_auxiliary_log_diagnostics_reads_only_current_attempt_append(self): + with TemporaryDirectory() as tmp: + path = Path(tmp) / "provider.log" + path.write_text( + "old error: Usage limit reached for 5 hour\n", + encoding="utf-8", + ) + offset = path.stat().st_size + path.write_text( + path.read_text(encoding="utf-8") + + "stream error: Usage limit reached for 5 hour\n" + + "Aborting non-transient provider quota retry\n", + encoding="utf-8", + ) + + diagnostics = dispatch.auxiliary_log_diagnostics(path, offset) + + self.assertEqual(len(diagnostics), 1) + self.assertIn("Usage limit reached", diagnostics[0]) + def test_output_validation_capability_rejection_is_provider_terminal(self): failure, evidence = dispatch.classify_failure_with_evidence( "no provider supports the required output validation capability" @@ -457,6 +477,37 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase): ) ) + def test_agent_end_diagnostic_ignores_historical_context_words(self): + failed = { + "type": "agent_end", + "willRetry": False, + "messages": [ + { + "role": "toolResult", + "content": [ + { + "type": "text", + "text": "context window token limit max_tokens model unavailable", + } + ], + }, + { + "role": "assistant", + "stopReason": "error", + "errorMessage": "502: provider_tunnel_error: recovery_failed", + }, + ], + } + + diagnostic = dispatch.terminal_diagnostic( + "pi", "stdout", json.dumps(failed) + ) + failure, evidence = dispatch.classify_failure_with_evidence(diagnostic or "") + + self.assertEqual(failure, "provider-connection") + self.assertIn("provider_tunnel_error", evidence or "") + self.assertNotIn("max_tokens", diagnostic or "") + def _invoke_fake_json_event( self, event: dict | list[dict],