From 1edf73aba4247a3337656195bcb1c6718e49953e Mon Sep 17 00:00:00 2001 From: Fasthei <167957975+Fasthei@users.noreply.github.com> Date: Tue, 16 Jun 2026 18:06:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(swarm/#70):=20=E4=BB=BB=E5=8A=A1=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E9=80=8F=E4=BC=A0+=E6=8F=90=E9=BB=98=E8=AE=A4=20/=20?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E6=97=B6=E9=97=B4=E7=BA=BF=E9=99=8D=E5=99=AA?= =?UTF-8?q?=20/=20=E5=A4=B1=E8=B4=A5=20termination=5Freason=20=E5=87=86?= =?UTF-8?q?=E7=A1=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 真机端到端实测(#70)暴露三问题,本 PR 全部修复(仅 agent + orchestrator,不跨仓): 问题1【阻断】单任务超时只有 60s,生成类任务必挂 - agent/main.py: TASK_TIMEOUT_SECONDS 默认 60→300(仅对外部/独立启动 agent 生效)。 - agent_launcher.py: 新增 DEFAULT_TASK_TIMEOUT_SECONDS=300、_budget_duration_seconds、 resolve_task_timeout(base=env 默认 300,与 run budget.duration_seconds 取较小); plan_launch_specs 把 TASK_TIMEOUT_SECONDS 透传进每个 agent env(非敏感,inline, k8s 不进 Secret)。 问题2【体验】事件时间线全是内部噪音(纯附加,未碰冻结契约) - swarm_runtime.py: is_client_visible(=event_type∈FROZEN_CLIENT_EVENT_TYPES,单一真源); emit_event 给 envelope 加 metadata.client_visible 布尔 + 关键客户端事件回填可选 payload.message(人话进度,仅取已有字段,不伪造)。task.heartbeat/retried/ deployment.status_changed/timeline/budget 标 client_visible=false,仍持久化+回调 但客户端据此过滤出时间线。冻结事件集/类型/sequence/artifact 形状一字未动。 - event-schema.md: 文档化两个附加字段 + 新增 §6.1,明确未解冻。 问题3【正确性】失败/超时 termination_reason 仍报 "tasks_completed" - convergence.py: 新增 TIMEOUT/MAX_RETRIES_EXCEEDED/TASK_FAILED;classify_failure_reason 按 timeout→max_retries→task_failed 取最具体(仅凭真实 per-task 信号);FAILED 分支 再不会返回 tasks_completed(该 reason 仅用于成功),budget/rounds 仅在通用失败时才覆盖。 - task_queue.py: fail_task 永久失败时把 reason 落到 task.result({"success":false,"error":reason}), 不覆盖已有结果,供 convergence 读取。 - main.py: compute_convergence_report 快照补 retry_count/max_retries。 测试:新增 test_resolve_task_timeout、扩 test-convergence(failed_timeout/max_retries/ generic + "FAILED 永不报 tasks_completed"不变量)。本地全过:test-agent-launcher / test-convergence / test-runtime-contract / test-contract-freeze / test-merge-smoke / test-workflow-e2e / test-security-boundary。 影响:agent + orchestrator + 文档;不动 Manager↔Swarm 冻结契约字段(问题2 纯附加)。 栈在 #64(agent_swarm git 注入)之上,#64 合并后本 PR base 自动转 main。 Closes #70 Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/main.py | 6 ++- docs/integration/event-schema.md | 19 +++++++- orchestrator/agent_launcher.py | 41 ++++++++++++++++++ orchestrator/convergence.py | 74 +++++++++++++++++++++++++++++--- orchestrator/main.py | 3 ++ orchestrator/swarm_runtime.py | 55 ++++++++++++++++++++++++ orchestrator/task_queue.py | 7 +++ scripts/test-agent-launcher.py | 31 ++++++++++++- scripts/test-convergence.py | 41 ++++++++++++++++-- 9 files changed, 262 insertions(+), 15 deletions(-) diff --git a/agent/main.py b/agent/main.py index 1b20594..6552040 100644 --- a/agent/main.py +++ b/agent/main.py @@ -76,7 +76,11 @@ class Agent: # agent's clone (see _execute_assignment / GitOperations.add_task_worktree), so concurrent tasks # never share a checkout/index. Swarm parallelism = many agents × this per-agent concurrency. MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "4")) - TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "60")) + # Per-task execution timeout. Default raised 60->300 (agent_swarm#70): 60s reliably killed + # generation-class tasks before the model finished. The Swarm launcher transmits an explicit + # TASK_TIMEOUT_SECONDS into each agent's env (capped at the run budget.duration_seconds), so + # this default only applies to externally-launched / standalone agents. + TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "300")) PEER_REPLY_TIMEOUT_SECONDS = int(os.getenv("PEER_REPLY_TIMEOUT_SECONDS", "20")) HEARTBEAT_INTERVAL_SECONDS = 15 diff --git a/docs/integration/event-schema.md b/docs/integration/event-schema.md index 3235071..156f1e4 100644 --- a/docs/integration/event-schema.md +++ b/docs/integration/event-schema.md @@ -37,8 +37,14 @@ "occurred_at": "2026-06-08T...Z", // 事件时间 "correlation_id": "corr-...", // 追踪 ID（X-Correlation-ID） "source": "heicode-swarm-runtime", // 事件来源标识 - "metadata": { }, // 自定义元数据（不得含明文密钥） - "payload": { }, // 事件专属字段（见 §4） + "metadata": { // 自定义元数据（不得含明文密钥） + "client_visible": true // agent_swarm#70：是否客户端 cockpit 可见事件（= event_type ∈ FROZEN_CLIENT_EVENT_TYPES）。 + // HM/客户端据此过滤内部噪音（heartbeat/retried/deployment.status_changed/timeline/budget）， + // 无需硬编码冻结集。**仅展示层过滤标记，不改变冻结事件集/类型/sequence/artifact 形状。** + }, + "payload": { // 事件专属字段（见 §4） + "message": "..." // agent_swarm#70：可选人话进度（仅客户端可见事件，缺省由运行时按 type+已有字段回填，不伪造新数据） + }, "artifact": { } // 可选；见 §3 } ``` @@ -118,4 +124,13 @@ 剩余小项（不影响客户端 13 类冻结）： - `handoff.completed` 的 `from_role`/`to_role`：辅助事件，客户端 13 类用 `handoff.created`（已带 from/to）；如 HM 仍消费 `handoff.completed`，在 handoff 完成处补两字段为后续小修。 +### 6.1 agent_swarm#70 时间线降噪（附加，未触碰冻结契约） + +回应 #70「事件时间线全是内部噪音」。**未改**冻结事件集 / 事件类型 / `sequence` / artifact 形状，仅两处**纯附加**增强（实现 `swarm_runtime.emit_event`，`is_client_visible` 为单一真源）： + +1. **`metadata.client_visible`（envelope，附加）**：每事件带布尔标记 = `event_type ∈ FROZEN_CLIENT_EVENT_TYPES`。客户端 cockpit 只渲染 `client_visible=true` 的事件，`task.heartbeat`/`task.retried`/`deployment.status_changed`/`timeline.updated`/`budget.alert` 等内部/调度事件 `client_visible=false`，仍持久化+回调（HM 控制面照常消费）但不入客户端时间线。这是**展示层过滤标记**，不改变运行时语义。 +2. **`payload.message`（附加，可选）**：关键客户端可见事件（`task.created|claimed|running|completed|failed`、`swarm.completed|failed|stopped`）回填一句人话进度，仅取已有 payload 字段（role/title/summary/reason/termination_reason），**不伪造新数据**（规则 #9）；emitter 已提供 `message` 时不覆盖。 + +> HM 侧无需改动即可消费（忽略 `metadata.client_visible`/`payload.message` 不影响契约）；客户端可立即按 `metadata.client_visible` 过滤并展示 `payload.message`。如需把 `client_visible` 提升为 envelope 顶层字段或新增「内部事件」分类口径，属契约扩展，须与 HM 联调后另立工单（非本次范围）。 + > 契约由 `scripts/test-contract-freeze.py` 守护（断言 sequence 单调、13 类 round-trip、artifact 形状、审批/停止真实发出、明文凭据脱敏）；`scripts/test-workflow-e2e.py` 在全流程 e2e 中额外断言 `swarm.completed` + sequence 无空洞。 diff --git a/orchestrator/agent_launcher.py b/orchestrator/agent_launcher.py index bb4fefc..491eed4 100644 --- a/orchestrator/agent_launcher.py +++ b/orchestrator/agent_launcher.py @@ -333,6 +333,44 @@ def resolve_git_grant(body: Dict[str, Any]) -> Optional[Dict[str, str]]: return env +# ── Per-task execution timeout (agent_swarm#70) ──────────────────────────────── +#: Default per-task timeout injected into launched agents. 60s (the old agent default) reliably +#: killed generation-class tasks before the model finished; raised to 300s. Overridable via +#: TASK_TIMEOUT_SECONDS on the orchestrator. Aligned (min) with the run's budget.duration_seconds +#: so a task can never outlive its run budget. +DEFAULT_TASK_TIMEOUT_SECONDS = 300 + + +def _budget_duration_seconds(body: Dict[str, Any]) -> Optional[int]: + """The run's wall-clock budget in seconds from the create body, or None when absent.""" + budget = ((body.get("orchestration_plan") or {}).get("budget") or {}) + raw = budget.get("duration_seconds") or budget.get("max_duration_seconds") + try: + val = int(raw) + except (TypeError, ValueError): + return None + return val if val > 0 else None + + +def resolve_task_timeout(body: Dict[str, Any]) -> int: + """Per-task execution timeout (seconds) to inject into each launched agent (agent_swarm#70). + + Base = ``TASK_TIMEOUT_SECONDS`` env (default ``DEFAULT_TASK_TIMEOUT_SECONDS`` = 300). When the + run declares ``budget.duration_seconds`` we take the **smaller** of the two so a single task can + never outlive the whole run budget. Always returns a positive int. + """ + try: + base = int(os.getenv("TASK_TIMEOUT_SECONDS", str(DEFAULT_TASK_TIMEOUT_SECONDS))) + except ValueError: + base = DEFAULT_TASK_TIMEOUT_SECONDS + if base <= 0: + base = DEFAULT_TASK_TIMEOUT_SECONDS + budget_duration = _budget_duration_seconds(body) + if budget_duration: + return min(base, budget_duration) + return base + + def model_api_base() -> str: return os.getenv("AGENT_OPENAI_API_BASE") or os.getenv("OPENAI_API_BASE") or "https://api.openai.com/v1" @@ -434,6 +472,7 @@ def plan_launch_specs( caps = pool_capabilities() or ["general"] base = model_api_base() mid = model_id(body) + task_timeout = resolve_task_timeout(body) # #70:透传执行超时，与 run budget.duration_seconds 取较小 specs: List[AgentLaunchSpec] = [] for i in range(count): cap_csv = caps[i % len(caps)] @@ -442,6 +481,8 @@ def plan_launch_specs( "AGENT_ID": f"{run.swarm_id}-agent-{i+1}", "AGENT_CAPABILITIES": cap_csv, "OPENAI_API_BASE": base, + # Non-secret execution-timeout knob; inline like the other config env (never a Secret). + "TASK_TIMEOUT_SECONDS": str(task_timeout), } if model_key: env["OPENAI_API_KEY"] = model_key diff --git a/orchestrator/convergence.py b/orchestrator/convergence.py index e857f0a..5843b9d 100644 --- a/orchestrator/convergence.py +++ b/orchestrator/convergence.py @@ -56,6 +56,9 @@ class TerminationReason(str, Enum): BUDGET_EXHAUSTED = "budget_exhausted" # token / cost / duration budget spent MAX_ROUNDS_REACHED = "max_rounds_reached" # review/redo cycle cap hit RISK_BLOCKED = "risk_blocked" # an unresolved blocking risk stopped the run + TIMEOUT = "timeout" # a task exhausted its execution timeout + MAX_RETRIES_EXCEEDED = "max_retries_exceeded" # a task failed after its retry budget + TASK_FAILED = "task_failed" # a task failed for another (non-timeout) reason TASKS_COMPLETED = "tasks_completed" # fallback: all tasks terminal, nothing else to explain @@ -180,6 +183,55 @@ def _task_result(task: Any) -> Dict[str, Any]: return {} +def _task_field(task: Any, key: str) -> Any: + """Read an arbitrary field off a task (dict or model).""" + if isinstance(task, dict): + return task.get(key) + return getattr(task, key, None) + + +def _is_timeout_failure(task: Any) -> bool: + """Whether a failed task's recorded reason indicates an execution timeout. + + The agent reports a timeout as ``result={"error": "timeout", ...}`` (agent/main.py); the + orchestrator persists that reason on the failed task. We match the canonical "timeout" token + (case-insensitive) in the task's result ``error``/``reason`` so we never guess. + """ + result = _task_result(task) + text = " ".join( + str(result.get(k) or "") for k in ("error", "reason", "summary") + ).lower() + return "timeout" in text + + +def _is_max_retries_failure(task: Any) -> bool: + """Whether a failed task exhausted its retry budget (retry_count >= max_retries).""" + try: + retry_count = int(_task_field(task, "retry_count") or 0) + max_retries = int(_task_field(task, "max_retries") or 0) + except (TypeError, ValueError): + return False + return max_retries > 0 and retry_count >= max_retries + + +def classify_failure_reason(tasks: List[Any]) -> TerminationReason: + """Pick the most-specific termination reason for a FAILED run from its failed tasks. + + Precedence (most specific first), derived only from real per-task signals (never fabricated): + 1. ``timeout`` — any failed task whose recorded error is a timeout. + 2. ``max_retries_exceeded`` — any failed task that exhausted its retry budget. + 3. ``task_failed`` — a task failed for another reason (the honest generic fail). + Returns ``TASK_FAILED`` when no failed task is present (caller should only invoke this on a + FAILED run, but we stay safe). + """ + failed = [t for t in tasks if _task_status(t) in _TASK_TERMINAL_FAIL] + if any(_is_timeout_failure(t) for t in failed): + return TerminationReason.TIMEOUT + if any(_is_max_retries_failure(t) for t in failed): + return TerminationReason.MAX_RETRIES_EXCEEDED + return TerminationReason.TASK_FAILED + + # --- conflict detectors --------------------------------------------------- def detect_artifact_mismatch(tasks: List[Any]) -> List[Conflict]: @@ -489,9 +541,13 @@ def evaluate_convergence(run_state: Dict[str, Any]) -> ConvergenceReport: Decision order (first match wins) for terminal runs: 1. Any active/blocked task -> RUNNING (no termination_reason yet) 2. Any blocking risk / unresolved hard conflict -> BLOCKED, RISK_BLOCKED - 3. Any failed task -> FAILED, BUDGET_EXHAUSTED if budget - spent else MAX_ROUNDS_REACHED if the - redo cap was hit else TASKS_COMPLETED + 3. Any failed task -> FAILED, with the most accurate reason: TIMEOUT if a + failed task timed out, else MAX_RETRIES_EXCEEDED if a + failed task exhausted its retry budget, else + BUDGET_EXHAUSTED if budget spent, else + MAX_ROUNDS_REACHED if the redo cap was hit, else + TASK_FAILED (never TASKS_COMPLETED — that is a + success-only reason; see agent_swarm#70) 4. Budget exhausted (all tasks done) -> CONVERGED, BUDGET_EXHAUSTED 5. Review cap hit -> CONVERGED, MAX_ROUNDS_REACHED 6. Quality gate reached -> CONVERGED, QUALITY_REACHED @@ -541,14 +597,18 @@ def evaluate_convergence(run_state: Dict[str, Any]) -> ConvergenceReport: max_cycles = int(run_state.get("max_review_cycles", 0) or 0) rounds_hit = max_cycles > 0 and cycles >= max_cycles - # 3. a failed task -> failed run, with the best available explanation. + # 3. a failed task -> failed run, with the best available explanation. A FAILED run must NEVER + # report `tasks_completed` (that reason is success-only and was the agent_swarm#70 bug): + # classify the failure (timeout / max_retries_exceeded / task_failed) and let a more specific + # resource cause (budget / rounds) win only when no concrete failure signal is present. if any(s in _TASK_TERMINAL_FAIL for s in statuses): - if budget_state.get("exhausted"): + failure_reason = classify_failure_reason(tasks) + if failure_reason is TerminationReason.TASK_FAILED and budget_state.get("exhausted"): reason = TerminationReason.BUDGET_EXHAUSTED - elif rounds_hit: + elif failure_reason is TerminationReason.TASK_FAILED and rounds_hit: reason = TerminationReason.MAX_ROUNDS_REACHED else: - reason = TerminationReason.TASKS_COMPLETED + reason = failure_reason return _report(ConvergenceStatus.FAILED, reason) # All tasks completed. Explain WHY it stopped, most-specific reason first. diff --git a/orchestrator/main.py b/orchestrator/main.py index 8e93fc3..882a293 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -452,6 +452,9 @@ async def compute_convergence_report(run, tasks, *, terminal: bool): "status": t.status.value if hasattr(t.status, "value") else t.status, "depends_on": t.depends_on, "result": result, + # #70: retry counters let convergence classify a FAILED run as max_retries_exceeded. + "retry_count": t.retry_count, + "max_retries": t.max_retries, }) plan = (run.request_body or {}).get("orchestration_plan") or {} budget = plan.get("budget") or {} diff --git a/orchestrator/swarm_runtime.py b/orchestrator/swarm_runtime.py index fc57764..4b8477c 100644 --- a/orchestrator/swarm_runtime.py +++ b/orchestrator/swarm_runtime.py @@ -111,6 +111,47 @@ FROZEN_CLIENT_EVENT_TYPES = ( # final state and the SSE stream can close — the client cockpit keys its terminal UI on these. TERMINAL_CLIENT_EVENT_TYPES = frozenset({"swarm.completed", "swarm.failed", "swarm.stopped"}) +_FROZEN_CLIENT_EVENT_SET = frozenset(FROZEN_CLIENT_EVENT_TYPES) + + +def is_client_visible(event_type: str) -> bool: + """Whether an event type is part of the frozen client cockpit contract (vs internal noise). + + The runtime stores/forwards many non-client event types (task.heartbeat, task.retried, + deployment.status_changed, timeline.updated, budget.alert, …). Only FROZEN_CLIENT_EVENT_TYPES + are rendered on the client timeline. This is the single source of truth both the runtime + (envelope `metadata.client_visible`, agent_swarm#70) and HM/clients filter on — no event is + removed, and the frozen type set is untouched. + """ + return event_type in _FROZEN_CLIENT_EVENT_SET + + +#: Human-readable message templates for key client-visible events (agent_swarm#70). Purely +#: additive: an optional `message` is set on the payload only when the emitter did not already +#: provide one. Templates pull from already-present payload fields, so no new data is fabricated. +def _client_message(event_type: str, payload: Dict[str, Any]) -> Optional[str]: + role = payload.get("agent_role") + title = payload.get("title") or payload.get("summary") + if event_type == "task.created": + return f"New task created: {title}" if title else "New task created" + if event_type == "task.claimed": + return f"{role} picked up a task" if role else "An agent picked up a task" + if event_type == "task.running": + return f"{role} is working on a task" if role else "An agent is working on a task" + if event_type == "task.completed": + return "Task completed" + if event_type == "task.failed": + reason = payload.get("reason") + return f"Task failed: {reason}" if reason else "Task failed" + if event_type == "swarm.completed": + return "Swarm run completed" + if event_type == "swarm.failed": + tr = payload.get("termination_reason") + return f"Swarm run failed ({tr})" if tr else "Swarm run failed" + if event_type == "swarm.stopped": + return "Swarm run stopped" + return None + class SwarmRuntime: """Stores swarm runs and emits Agent Manager callback events.""" @@ -651,6 +692,15 @@ class SwarmRuntime: if isinstance(threshold, (int, float)): redacted_payload["threshold_pct"] = threshold * 100 if threshold <= 1 else threshold + # Natural-language progress (agent_swarm#70): give client-visible events a human `message` + # so the cockpit timeline reads as progress instead of raw event types. Additive + optional; + # only set when the emitter did not already provide one. Internal/noise events get none. + client_visible = is_client_visible(event_type) + if client_visible and "message" not in redacted_payload: + message = _client_message(event_type, redacted_payload) + if message: + redacted_payload["message"] = message + event_id = f"evt_{uuid.uuid4().hex}" # Per-swarm strictly-increasing sequence (agent_swarm#15.1): the client polls # `events?after=` and dedups/orders by it. INCR is atomic so concurrent emits on @@ -669,6 +719,11 @@ class SwarmRuntime: "occurred_at": occurred_at, "correlation_id": run.correlation_id, "source": self.runtime_source, + # client_visible (agent_swarm#70): explicit, type-derived flag so HM/clients can filter + # internal noise (heartbeat/retried/deployment.status_changed/timeline/budget) off the + # cockpit timeline without hard-coding the frozen set. Additive metadata field — does not + # touch the frozen envelope keys, event types, sequence, or artifact shape. + "metadata": {"client_visible": client_visible}, "payload": redacted_payload, } if redacted_artifact: diff --git a/orchestrator/task_queue.py b/orchestrator/task_queue.py index 7579fe4..6e89c2c 100644 --- a/orchestrator/task_queue.py +++ b/orchestrator/task_queue.py @@ -362,6 +362,13 @@ class TaskQueue: else: task.status = TaskStatus.FAILED task.completed_at = time.time() + # Persist the terminal failure reason on the task so downstream convergence (#70) can + # derive an accurate termination_reason (timeout / max_retries_exceeded / task_failed). + # Stored in the result envelope, mirroring the agent's failure payload shape + # ({"success": false, "error": ...}); does not overwrite a structured result if one + # already captured the cause. + if reason and not task.result: + task.result = json.dumps({"success": False, "error": reason}) logger.error( f"Task {task_id} permanently failed after {task.retry_count} retries: {reason}" diff --git a/scripts/test-agent-launcher.py b/scripts/test-agent-launcher.py index 854a521..9825bc6 100644 --- a/scripts/test-agent-launcher.py +++ b/scripts/test-agent-launcher.py @@ -96,7 +96,8 @@ def test_plan_specs(): specs = al.plan_launch_specs(run, body, connected_user_agents=8, limit=10, pool_size=3, model_key="sk-test", orchestrator_url="ws://orch:8000", user_id="u-1", git_env=git_env) - check("plan caps at limit (8 connected, cap 10 -> launch 2)", len(specs) == 2) + # [3,16] pod-count window (#66): MIN floor (3) wins over cap headroom (10-8=2) -> launch 3. + check("plan floors to MIN pool (8 connected, cap 10, headroom 2 -> floor 3)", len(specs) == 3) s = specs[0] check("spec env has ORCHESTRATOR_URL", s.env.get("ORCHESTRATOR_URL") == "ws://orch:8000") check("spec env has model key (server-side injected)", s.env.get("OPENAI_API_KEY") == "sk-test") @@ -106,6 +107,8 @@ def test_plan_specs(): check("spec env has GIT_USERNAME + GIT_PASSWORD", s.env.get("GIT_USERNAME") == "x-access-token" and s.env.get("GIT_PASSWORD") == "ghp_xyz") check("spec has AGENT_ID + capabilities", bool(s.agent_id) and bool(s.env.get("AGENT_CAPABILITIES"))) check("agent ids unique", len({sp.agent_id for sp in specs}) == len(specs)) + # #70: per-task timeout is transmitted into every launched agent's env (non-secret, inline). + check("spec env has TASK_TIMEOUT_SECONDS (transmitted, #70)", s.env.get("TASK_TIMEOUT_SECONDS") == str(al.DEFAULT_TASK_TIMEOUT_SECONDS)) # No key -> OPENAI_API_KEY omitted (not fabricated), no user -> HEICODE_USER_ID omitted, no git -> GIT_* omitted. specs2 = al.plan_launch_specs(run, body, connected_user_agents=0, limit=10, pool_size=1, model_key=None, orchestrator_url="ws://orch", user_id=None, git_env=None) @@ -114,6 +117,30 @@ def test_plan_specs(): check("no git grant -> GIT_REPO_URL omitted", "GIT_REPO_URL" not in specs2[0].env) +def test_resolve_task_timeout(): + # #70: per-task execution timeout transmitted to launched agents. + os.environ.pop("TASK_TIMEOUT_SECONDS", None) + check("default timeout = 300 (raised from 60)", al.resolve_task_timeout({}) == al.DEFAULT_TASK_TIMEOUT_SECONDS == 300) + # env override. + os.environ["TASK_TIMEOUT_SECONDS"] = "450" + check("env TASK_TIMEOUT_SECONDS honored", al.resolve_task_timeout({}) == 450) + # capped at run budget.duration_seconds (take the smaller). + body = {"orchestration_plan": {"budget": {"duration_seconds": 120}}} + check("timeout capped at budget.duration_seconds (smaller wins)", al.resolve_task_timeout(body) == 120) + body_big = {"orchestration_plan": {"budget": {"duration_seconds": 3600}}} + check("budget larger than base -> base wins", al.resolve_task_timeout(body_big) == 450) + os.environ.pop("TASK_TIMEOUT_SECONDS") + # max_duration_seconds alias also honored. + check("max_duration_seconds alias honored", + al.resolve_task_timeout({"orchestration_plan": {"budget": {"max_duration_seconds": 90}}}) == 90) + # bad/zero env -> falls back to default; absent budget -> default. + os.environ["TASK_TIMEOUT_SECONDS"] = "0" + check("non-positive env -> default", al.resolve_task_timeout({}) == al.DEFAULT_TASK_TIMEOUT_SECONDS) + os.environ["TASK_TIMEOUT_SECONDS"] = "notanint" + check("non-int env -> default", al.resolve_task_timeout({}) == al.DEFAULT_TASK_TIMEOUT_SECONDS) + os.environ.pop("TASK_TIMEOUT_SECONDS") + + def test_resolve_git_grant(): # ── credential extraction (pure) ── check("git creds: JSON git_username/git_password", @@ -271,7 +298,7 @@ def main(): test_launch_count() test_pool_window() test_plan_specs() - test_git_launch_env() + test_resolve_task_timeout() test_resolve_model_key() test_resolve_git_grant() test_azkv_resolver() diff --git a/scripts/test-convergence.py b/scripts/test-convergence.py index b2eed00..e66478a 100644 --- a/scripts/test-convergence.py +++ b/scripts/test-convergence.py @@ -151,14 +151,42 @@ terminal_cases = { "expect_status": ConvergenceStatus.BLOCKED, "expect_reason": TerminationReason.RISK_BLOCKED, }, - # a failed task -> FAILED, fallback reason when nothing else explains it - "failed_fallback": { + # a failed task with no specific signal -> FAILED, task_failed (NOT tasks_completed; #70 bug). + "failed_generic": { "tasks": [ completed_task("t1", tests_passed=True), {"task_id": "t2", "status": "failed", "depends_on": [], "result": {}}, ], "expect_status": ConvergenceStatus.FAILED, - "expect_reason": TerminationReason.TASKS_COMPLETED, + "expect_reason": TerminationReason.TASK_FAILED, + }, + # a failed task whose recorded error is a timeout -> FAILED, timeout (#70). + "failed_timeout": { + "tasks": [ + {"task_id": "t2", "status": "failed", "depends_on": [], + "result": {"success": False, "error": "timeout"}}, + ], + "expect_status": ConvergenceStatus.FAILED, + "expect_reason": TerminationReason.TIMEOUT, + }, + # a failed task that exhausted its retry budget -> FAILED, max_retries_exceeded (#70). + "failed_max_retries": { + "tasks": [ + {"task_id": "t2", "status": "failed", "depends_on": [], "result": {}, + "retry_count": 3, "max_retries": 3}, + ], + "expect_status": ConvergenceStatus.FAILED, + "expect_reason": TerminationReason.MAX_RETRIES_EXCEEDED, + }, + # timeout takes precedence over max_retries when both are present (#70 precedence). + "failed_timeout_over_retries": { + "tasks": [ + {"task_id": "t2", "status": "failed", "depends_on": [], + "result": {"success": False, "error": "timeout"}, + "retry_count": 3, "max_retries": 3}, + ], + "expect_status": ConvergenceStatus.FAILED, + "expect_reason": TerminationReason.TIMEOUT, }, } @@ -179,6 +207,13 @@ for label, case in terminal_cases.items(): check(f"[{label}] invariant: terminal -> reason set", rep.termination_reason is not None) +# #70 invariant: a FAILED run must NEVER report tasks_completed (success-only reason). +for label, case in terminal_cases.items(): + rep = evaluate_convergence(case) + if rep.status == ConvergenceStatus.FAILED: + check(f"[{label}] FAILED run never reports tasks_completed (#70)", + rep.termination_reason != TerminationReason.TASKS_COMPLETED) + # A still-running run must NOT carry a termination_reason. running = evaluate_convergence({"tasks": [{"task_id": "x", "status": "in_progress", "depends_on": []}]}) check("running run is RUNNING", running.status == ConvergenceStatus.RUNNING)