Feat/v2

.agent-role-contracts.yaml

@@ -68,13 +68,17 @@ roles:
   quality-runner:
     prompt_path: .agents/prompts/quality-runner.md
     output_contract: schemas/roles/quality-runner.schema.json
+    execution_kind: harness_stage
+    llm_invocation: false
     expected_artifacts:
       - quality.json
     artifact_schemas:
       quality.json: schemas/roles/quality-runner.schema.json
   security-agent:
     prompt_path: .agents/prompts/security-agent.md
     output_contract: schemas/roles/security-agent.schema.json
+    execution_kind: harness_stage
+    llm_invocation: false
     expected_artifacts:
       - security.json
     artifact_schemas:
@@ -117,6 +121,8 @@ roles:
   orchestrator:
     prompt_path: .agents/prompts/orchestrator.md
     output_contract: schemas/roles/orchestrator.schema.json
+    execution_kind: harness_stage
+    llm_invocation: false
     expected_artifacts:
       - verdict.json
     artifact_schemas:

.agent-runtime.yaml

@@ -7,7 +7,18 @@ runtime:
   transport: local_subscription
   api_required: false
   model_router: false
-  model: gpt-5.6-sol
-  reasoning_effort: high
-  service_tier: fast
+  default_execution_profile: balanced
+  execution_profiles:
+    complex:
+      model: gpt-5.6-sol
+      reasoning_effort: high
+      service_tier: fast
+    balanced:
+      model: gpt-5.6-terra
+      reasoning_effort: medium
+      service_tier: fast
+    economy:
+      model: gpt-5.6-luna
+      reasoning_effort: low
+      service_tier: fast
   require_account_type: chatgpt

.agents/prompts/implementation-agent.md

@@ -100,8 +100,15 @@ At completion, return `implementation.json` with:
 
 ## Verification handoff
 Do not claim the task is complete. After implementation, hand off to:
-1. Test Generator Agent, if tests are incomplete
-2. Quality Runner Agent
-3. Security Agent
-4. Reviewer Agent
-5. Orchestrator Agent
+1. Test Generator Agent when code changed and test work is required
+2. deterministic Quality Runner
+3. deterministic Security Agent
+4. optional impact-specific verifiers selected from changed files and risk
+5. Reviewer Agent, model-backed only for code, UI, risk-bearing, or large changes
+6. deterministic Orchestrator Agent
+
+## Independent background work
+
+Keep blocking compile/test failures in the current run's bounded repair loop. Only when a newly discovered repair or investigation is independent enough to run in a separate worktree may you propose up to three top-level `child_tasks` in the structured role result. The Harness, not the model, decides whether to enqueue them.
+
+Each proposal must stay inside the current repository and include `task_id`, `goal`, `repository`, `relation`, `dependency_mode`, `spawn_reason`, a narrow `allowed_paths` list, `max_tokens` no greater than 40000, and `max_duration_seconds` no greater than 900. Use `blocking` when the parent cannot pass its next gate without the result; otherwise use `non_blocking`. Do not propose a child merely to repeat the current role, bypass a failed gate, change protected scope, publish, merge, deploy, or access another repository. Return an empty `child_tasks` list when no genuinely independent work exists.

ai_harness/branch_conflicts.py

@@ -0,0 +1,89 @@
+"""Deterministic overlap analysis for concurrent task branches."""
+
+from __future__ import annotations
+
+import subprocess
+from itertools import combinations
+from pathlib import Path
+from typing import Any
+
+
+TERMINAL_STATUSES = {"completed", "cancelled", "dead_letter", "failed"}
+
+
+def checkout_changed_paths(checkout: Path) -> set[str]:
+    if not checkout.is_dir():
+        return set()
+    paths: set[str] = set()
+    for command in (
+        ["git", "diff", "--name-only", "HEAD"],
+        ["git", "ls-files", "--others", "--exclude-standard"],
+    ):
+        try:
+            result = subprocess.run(
+                command,
+                cwd=checkout,
+                text=True,
+                capture_output=True,
+                check=False,
+                timeout=5,
+            )
+        except (OSError, subprocess.TimeoutExpired):
+            continue
+        if result.returncode == 0:
+            paths.update(line.strip() for line in result.stdout.splitlines() if line.strip())
+    return paths
+
+
+def analyze_branch_conflicts(
+    queue_items: list[dict[str, Any]],
+    runs_by_id: dict[str, dict[str, Any]],
+) -> list[dict[str, Any]]:
+    candidates: list[dict[str, Any]] = []
+    for task in queue_items:
+        if str(task.get("status", "")) in TERMINAL_STATUSES:
+            continue
+        payload = task.get("payload", {})
+        if not isinstance(payload, dict):
+            continue
+        run = runs_by_id.get(str(task.get("run_id", "")), {})
+        repository = str(payload.get("repository") or run.get("repository") or "")
+        branch = str(payload.get("task_branch") or payload.get("branch") or run.get("branch") or "")
+        checkout = str(run.get("checkout_path") or payload.get("checkout_path") or "")
+        if not repository or not branch or not checkout:
+            continue
+        candidates.append(
+            {
+                "queue_task_id": int(task.get("id", 0) or 0),
+                "run_id": str(task.get("run_id", "")),
+                "task_id": str(payload.get("task_id", "")),
+                "repository": repository,
+                "branch": branch,
+                "paths": checkout_changed_paths(Path(checkout)),
+            }
+        )
+    conflicts: list[dict[str, Any]] = []
+    for left, right in combinations(candidates, 2):
+        if left["repository"] != right["repository"] or left["branch"] == right["branch"]:
+            continue
+        overlap = sorted(left["paths"] & right["paths"])
+        if not overlap:
+            continue
+        first, second = sorted((left, right), key=lambda item: item["queue_task_id"])
+        conflicts.append(
+            {
+                "repository": left["repository"],
+                "run_ids": [left["run_id"], right["run_id"]],
+                "task_ids": [left["task_id"], right["task_id"]],
+                "branches": [left["branch"], right["branch"]],
+                "overlapping_paths": overlap[:20],
+                "overlap_count": len(overlap),
+                "recommended_first_run_id": first["run_id"],
+                "recommended_rebase_run_id": second["run_id"],
+                "recommendation": (
+                    f"publish {first['branch']} first, then rebase {second['branch']} "
+                    "before its final verification"
+                ),
+            }
+        )
+    return conflicts