sync: from iop-s2 v1.1.204

agent-ops/.version

@@ -1 +1 @@
-1.1.203
+1.1.204

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"
         }

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"
+        event_type in {"error", "fatal", "request.failed", "turn.failed", "rate_limit_event"}
-        or 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": [
+        "auxiliary_logs": auxiliary_logs,
-            str(item).format_map(
+        "auxiliary_log_offsets": auxiliary_log_offsets,
-                {
-                    "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", [])
-        ],
         "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

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],