fix(agent-ops): 정지한 모델 세션을 자동 복구한다
This commit is contained in:
parent
cdef6be96a
commit
2802da5ad0
4 changed files with 78 additions and 13 deletions
|
|
@ -183,7 +183,7 @@
|
|||
"--model",
|
||||
"iop-glm/glm-5.2",
|
||||
"--variant",
|
||||
"max",
|
||||
"high",
|
||||
"--auto",
|
||||
"{prompt}"
|
||||
],
|
||||
|
|
|
|||
|
|
@ -187,9 +187,8 @@ PROVIDER_TRANSPORT_FAILURES = frozenset(
|
|||
{"provider-connection", "provider-stream-disconnect"}
|
||||
)
|
||||
FAILURE_EVIDENCE_LIMIT = 2000
|
||||
# Used only to reject a stale locator whose dispatcher and agent PIDs are both
|
||||
# gone. A live process is inspected after silence; it is never killed solely by
|
||||
# this fallback clock.
|
||||
# Used to reject stale locators and to bound a live model response that stops
|
||||
# producing both stream and native-session progress outside tool execution.
|
||||
RUNTIME_FAILURE_PATTERNS = {
|
||||
"context-limit": [
|
||||
r"context (?:length|window)", r"maximum context", r"prompt is too long",
|
||||
|
|
@ -3349,6 +3348,7 @@ async def invoke(
|
|||
diagnostics: list[str] = []
|
||||
diagnostic_origins: list[str] = []
|
||||
control_violation: str | None = None
|
||||
session_stall_seconds: float | None = None
|
||||
terminal_success_contract = spec.runtime.get("terminal_success")
|
||||
terminal_success_seen = False
|
||||
try:
|
||||
|
|
@ -3527,8 +3527,9 @@ async def invoke(
|
|||
spec.native_resume
|
||||
and not is_native_tool_execution
|
||||
and native_inactive_seconds >= MODEL_RESPONSE_STALL_SECONDS
|
||||
and "native_silence_inspection" not in record
|
||||
and session_stall_seconds is None
|
||||
):
|
||||
session_stall_seconds = native_inactive_seconds
|
||||
inspection = {
|
||||
"at": now_iso(),
|
||||
"silence_seconds": round(native_inactive_seconds, 3),
|
||||
|
|
@ -3537,13 +3538,14 @@ async def invoke(
|
|||
record["native_silence_inspection"] = inspection
|
||||
diagnostic = (
|
||||
f"native-session {native_phase} stream produced no update for "
|
||||
f"{native_inactive_seconds:.1f}s; recorded stream tail for inspection "
|
||||
"without terminating the model process"
|
||||
f"{native_inactive_seconds:.1f}s; terminating the stalled model "
|
||||
"process for native-session recovery"
|
||||
)
|
||||
heartbeat_log.write(f"[silence-inspection] {diagnostic}\n")
|
||||
heartbeat_log.flush()
|
||||
persist_locator_record()
|
||||
attempt_event(prefix, f"모델응답점검: {diagnostic}")
|
||||
attempt_event(prefix, f"모델응답정지: {diagnostic}")
|
||||
await terminate_process_group(process)
|
||||
non_native_inactive_seconds = loop.time() - max(
|
||||
last_native_progress_at, last_stream_progress_at
|
||||
)
|
||||
|
|
@ -3551,8 +3553,9 @@ async def invoke(
|
|||
not spec.native_resume
|
||||
and non_native_inactive_seconds
|
||||
>= MODEL_RESPONSE_STALL_SECONDS
|
||||
and "stream_silence_inspection" not in record
|
||||
and session_stall_seconds is None
|
||||
):
|
||||
session_stall_seconds = non_native_inactive_seconds
|
||||
inspection = {
|
||||
"at": now_iso(),
|
||||
"silence_seconds": round(non_native_inactive_seconds, 3),
|
||||
|
|
@ -3561,13 +3564,14 @@ async def invoke(
|
|||
record["stream_silence_inspection"] = inspection
|
||||
diagnostic = (
|
||||
f"{spec.cli} emitted no stream output or native-session event for "
|
||||
f"{non_native_inactive_seconds:.1f}s; recorded stream tail for inspection "
|
||||
"without terminating the model process"
|
||||
f"{non_native_inactive_seconds:.1f}s; terminating the stalled model "
|
||||
"process for retry recovery"
|
||||
)
|
||||
heartbeat_log.write(f"[silence-inspection] {diagnostic}\n")
|
||||
heartbeat_log.flush()
|
||||
persist_locator_record()
|
||||
attempt_event(prefix, f"모델응답점검: {diagnostic}")
|
||||
attempt_event(prefix, f"모델응답정지: {diagnostic}")
|
||||
await terminate_process_group(process)
|
||||
heartbeat = (
|
||||
f"작업중... locator={locator_path} "
|
||||
f"native_session={record.get('native_session_path') or 'none'} "
|
||||
|
|
@ -3703,6 +3707,11 @@ async def invoke(
|
|||
if diagnostic_origins[index] == failure_evidence_source:
|
||||
failure_evidence = diagnostics[index]
|
||||
break
|
||||
elif session_stall_seconds is not None:
|
||||
failure_class = "session-stall"
|
||||
failure_source = "dispatcher-stall-timeout"
|
||||
record["session_stall_seconds"] = round(session_stall_seconds, 3)
|
||||
record["termination_initiator"] = "dispatcher"
|
||||
elif return_code != 0 and termination is not None:
|
||||
failure_class = "process-terminated"
|
||||
failure_source = "process-termination"
|
||||
|
|
|
|||
|
|
@ -443,6 +443,61 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
|
|||
self.assertEqual(record["status"], "succeeded")
|
||||
self.assertIn("succeeded:0", work_log)
|
||||
|
||||
def test_silent_native_session_is_terminated_and_classified_as_stall(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
plan = write_plan(root)
|
||||
task = task_from_plan(root, plan)
|
||||
runner = root / "silent_runner.py"
|
||||
runner.write_text(
|
||||
"import time\n"
|
||||
"time.sleep(30)\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
agent = dispatch.AgentSpec(
|
||||
"silent-runner",
|
||||
"silent-model",
|
||||
"silent-runner/silent-model",
|
||||
native_resume=True,
|
||||
target_id="silent-target",
|
||||
runtime={
|
||||
"command": [sys.executable, str(runner)],
|
||||
"native_session_monitor": True,
|
||||
"session_path": "sessions/{session_id}.jsonl",
|
||||
},
|
||||
)
|
||||
with (
|
||||
mock.patch.dict(
|
||||
os.environ,
|
||||
{"XDG_STATE_HOME": str(root / "state")},
|
||||
),
|
||||
mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
|
||||
mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
|
||||
):
|
||||
store = dispatch.StateStore(root)
|
||||
try:
|
||||
return_code, failure, locator = asyncio.run(
|
||||
dispatch.invoke(
|
||||
root,
|
||||
store,
|
||||
task,
|
||||
"worker",
|
||||
agent,
|
||||
"fake prompt",
|
||||
)
|
||||
)
|
||||
record = json.loads(locator.read_text(encoding="utf-8"))
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
self.assertLess(return_code, 0)
|
||||
self.assertEqual(failure, "session-stall")
|
||||
self.assertEqual(record["status"], "failed")
|
||||
self.assertEqual(record["failure_source"], "dispatcher-stall-timeout")
|
||||
self.assertEqual(record["termination_initiator"], "dispatcher")
|
||||
self.assertGreaterEqual(record["session_stall_seconds"], 0.05)
|
||||
self.assertIn("native_silence_inspection", record)
|
||||
|
||||
def test_catalog_source_is_in_runtime_audit_evidence(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
|
|
|
|||
|
|
@ -173,7 +173,8 @@ class SelectorTests(unittest.TestCase):
|
|||
self.assertEqual(agy.runtime["auxiliary_logs"], ["{attempt_dir}/agy-cli.log"])
|
||||
opencode = catalog.targets["opencode-glm-max"]
|
||||
self.assertIn("iop-glm/glm-5.2", opencode.runtime["command"])
|
||||
self.assertIn("max", opencode.runtime["command"])
|
||||
self.assertIn("high", opencode.runtime["command"])
|
||||
self.assertNotIn("max", opencode.runtime["command"])
|
||||
claude = catalog.targets["claude-opus-xhigh"]
|
||||
self.assertEqual((claude.agent, claude.model), ("claude", "claude-opus-5"))
|
||||
terra = catalog.targets["codex-terra-high"]
|
||||
|
|
|
|||
Loading…
Reference in a new issue