diff --git a/skills/swarm/scripts/cloud_dispatch.py b/skills/swarm/scripts/cloud_dispatch.py
index 76a6c80..8097899 100644
--- a/skills/swarm/scripts/cloud_dispatch.py
+++ b/skills/swarm/scripts/cloud_dispatch.py
@@ -32,6 +32,12 @@
is_partner_model = None
PARTNER_MODEL_ALIASES = {}
+try:
+ from worker.execution import LIMIT_ENV, TERMINAL_STATES, resolve_execution_limits
+except ImportError:
+ from skills.swarm.scripts.worker.execution import LIMIT_ENV, TERMINAL_STATES, resolve_execution_limits
+
+
TIER3_DEDICATED_MODELS = {
"glm-5-3-flash",
"glm-5.3-flash",
@@ -370,6 +376,14 @@ def _flush():
for line in log_output.splitlines():
trimmed = line.strip()
if trimmed.startswith("[STATUS:"):
+ if not current_status and current_evidence:
+ current_status = trimmed.split("[STATUS:")[1].split("]")[0].strip()
+ _flush()
+ current_status = ""
+ current_evidence = []
+ current_summary = []
+ mode = None
+ continue
_flush()
current_status = trimmed.split("[STATUS:")[1].split("]")[0].strip()
current_evidence = []
@@ -1174,6 +1188,19 @@ def resolve_swarm_model(
)
+def failed_task_count(report_items: List[Dict[str, Any]], task_count: int) -> int:
+ reports = {}
+ invalid = 0
+ for item in report_items:
+ index = item.get("task_index")
+ if type(index) is not int or not 0 <= index < task_count or index in reports:
+ invalid += 1
+ continue
+ status = item.get("status")
+ reports[index] = status if isinstance(status, str) and status in TERMINAL_STATES else "UNKNOWN"
+ return invalid + sum(reports.get(index) != "PASS" for index in range(task_count))
+
+
def print_candidate_patches_table(
harvested_patches: List[Dict[str, Any]],
baseline_score: Optional[float] = None,
@@ -1364,6 +1391,7 @@ def main() -> None:
return
runtime_cfg = find_runtime_config() or {}
+
job_name = args.job_name or runtime_cfg.get("job_name") or "agystack-swarm-worker"
region = args.region or runtime_cfg.get("region") or "us-central1"
project = args.project or runtime_cfg.get("project_id")
@@ -1417,7 +1445,15 @@ def main() -> None:
print(f"Error: Failed to harvest GCS results for session {session_id}: {exc}", file=sys.stderr)
sys.exit(1)
- print_candidate_patches_table(harvested_patches, baseline_score=args.baseline_score)
+ failed_count = failed_task_count(harvested_patches, args.tasks) if args.tasks and args.tasks > 0 else 1
+ print_candidate_patches_table(
+ harvested_patches, baseline_score=args.baseline_score, failed_count=failed_count,
+ allow_partial=args.ignore_partial_failures,
+ )
+ if not args.tasks:
+ print("Complete task coverage is unknown. Supply --tasks to verify the expected count.")
+ if failed_count:
+ sys.exit(1)
return
if args.wait_execution:
@@ -1477,8 +1513,13 @@ def main() -> None:
except Exception as exc:
print(f"Warning: Failed to harvest GCS results: {exc}", file=sys.stderr)
+ expected_count = args.tasks
+ if isinstance(desc_data, dict):
+ expected_count = desc_data.get("spec", {}).get("taskCount", desc_data.get("taskCount", expected_count))
+ failed_cnt = failed_task_count(harvested_patches, expected_count) if type(expected_count) is int and expected_count > 0 else 1
+ if not exec_succeeded:
+ failed_cnt = max(1, failed_cnt)
if harvested_patches:
- failed_cnt = 0 if exec_succeeded else 1
print_candidate_patches_table(
harvested_patches,
baseline_score=args.baseline_score,
@@ -1487,10 +1528,36 @@ def main() -> None:
allow_partial=getattr(args, "ignore_partial_failures", False),
)
- if not exec_succeeded:
+ if not exec_succeeded or failed_cnt:
+ print("Execution lacks complete passing task status coverage.", file=sys.stderr)
sys.exit(1)
return
+ tasks_list: List[Any] = []
+ sys.path.insert(0, str(Path(__file__).parent))
+
+ if args.seed_lottery:
+ from manifest_generator import generate_seed_lottery
+ tasks_list = generate_seed_lottery(seeds_count=args.seed_lottery)
+ elif args.manifest:
+ try:
+ tasks_list = load_manifest(args.manifest)
+ except Exception as exc:
+ print(f"Error: Failed to load manifest: {exc}", file=sys.stderr)
+ sys.exit(1)
+
+ try:
+ resolve_execution_limits()
+ for task in tasks_list:
+ if isinstance(task, dict):
+ resolve_execution_limits(task.get("execution_limits", {}), environ={})
+ except ValueError as exc:
+ parser.error(str(exc))
+ task_count = args.tasks or (len(tasks_list) if tasks_list else 1)
+ if tasks_list and task_count != len(tasks_list):
+ parser.error("--tasks must match the manifest task count")
+ manifest_serialized = json.dumps(tasks_list) if tasks_list else ""
+
repo_url = args.repo or get_git_remote_url()
gh_token = get_gh_token()
gemini_api_key = os.environ.get("GEMINI_API_KEY", "").strip()
@@ -1563,29 +1630,15 @@ def main() -> None:
sys.exit(1)
if not gh_token and "GH_TOKEN" not in active_secret_keys and "gh_token" not in active_secret_keys:
- print("Error: GH_TOKEN could not be resolved from environment or `gh auth token`", file=sys.stderr)
- sys.exit(1)
+ if not args.dry_run:
+ print("Error: GH_TOKEN could not be resolved from environment or `gh auth token`", file=sys.stderr)
+ sys.exit(1)
+ gh_token = "DRY_RUN_GH_TOKEN"
if not use_vertex and not gemini_api_key and not args.dry_run and "GEMINI_API_KEY" not in active_secret_keys and "gemini_api_key" not in active_secret_keys:
print("Error: GEMINI_API_KEY environment variable is required when not in Vertex AI mode.", file=sys.stderr)
sys.exit(1)
- tasks_list: List[Any] = []
- sys.path.insert(0, str(Path(__file__).parent))
-
- if args.seed_lottery:
- from manifest_generator import generate_seed_lottery
- tasks_list = generate_seed_lottery(seeds_count=args.seed_lottery)
- elif args.manifest:
- try:
- tasks_list = load_manifest(args.manifest)
- except Exception as exc:
- print(f"Error: Failed to load manifest: {exc}", file=sys.stderr)
- sys.exit(1)
-
- task_count = args.tasks or (len(tasks_list) if tasks_list else 1)
- manifest_serialized = json.dumps(tasks_list) if tasks_list else ""
-
env_vars = {
"REPO_URL": repo_url,
}
@@ -1607,6 +1660,9 @@ def main() -> None:
session_id = args.session_id or f"swarm-{int(time.time())}"
env_vars["SWARM_SESSION_ID"] = session_id
env_vars["SESSION_ID"] = session_id
+ for name, env_key in LIMIT_ENV.items():
+ if env_key in os.environ:
+ env_vars[env_key] = os.environ[env_key]
if args.orchestrator_timeout:
env_vars["ORCHESTRATOR_TIMEOUT"] = str(args.orchestrator_timeout)
if args.base_branch:
@@ -1881,7 +1937,7 @@ def main() -> None:
print(f"Total: {len(report_items)} | PASS: {pass_count} | ISSUES: {issues_count} | BLOCKED: {blocked_count}")
print("=" * 80)
- failed_tasks_count = issues_count + blocked_count
+ failed_tasks_count = failed_task_count(report_items, task_count)
if not exec_succeeded and failed_tasks_count == 0:
failed_tasks_count = 1
@@ -1901,7 +1957,7 @@ def main() -> None:
cleanup_session(session_id=session_id, bucket_name=gcs_bucket)
- if not exec_succeeded:
+ if not exec_succeeded or failed_tasks_count > 0:
sys.exit(1)
except subprocess.CalledProcessError as exc:
diff --git a/skills/swarm/scripts/cloud_worker.py b/skills/swarm/scripts/cloud_worker.py
index 305d59b..3f61849 100644
--- a/skills/swarm/scripts/cloud_worker.py
+++ b/skills/swarm/scripts/cloud_worker.py
@@ -10,7 +10,6 @@
import json
import os
-import re
import shutil
import subprocess
import sys
@@ -34,7 +33,7 @@
1. Implement the requested task directly and surgically in the workspace.
2. Maintain high engineering rigor: run test suites and verification commands before completing.
3. Keep diffs minimal and clean. Avoid narrating comments or speculative code.
-4. Conclude with a clear summary and status ([STATUS: PASS] or [STATUS: ISSUES]).
+4. End your final response with exactly one standalone unquoted status line outside code fences, [STATUS: PASS], [STATUS: ISSUES], or [STATUS: BLOCKED].
"""
@@ -169,20 +168,10 @@ def emit_agent_response(
)
-CREDENTIAL_PATTERNS = [
- re.compile(r"ghp_[A-Za-z0-9_]{36,}"),
- re.compile(r"AIza[0-9A-Za-z-_]{35}"),
- re.compile(r"ya29\.[0-9A-Za-z-_]+"),
- re.compile(r"Bearer [A-Za-z0-9_\-\.]+"),
-]
-
-
-def scrub_credentials(text: str) -> str:
- if not text:
- return text
- for pattern in CREDENTIAL_PATTERNS:
- text = pattern.sub("[REDACTED]", text)
- return text
+try:
+ from worker.execution import CREDENTIAL_PATTERNS as CREDENTIAL_PATTERNS, scrub_credentials
+except ImportError:
+ from skills.swarm.scripts.worker.execution import CREDENTIAL_PATTERNS as CREDENTIAL_PATTERNS, scrub_credentials
class TokenShield:
@@ -701,14 +690,10 @@ def parse_porcelain_status(stdout: str) -> List[str]:
return modified_files
-def evaluate_agent_status(agent_text: str) -> Tuple[str, str]:
- match = re.search(r"\[STATUS:\s*(PASS|ISSUES|BLOCKED|DEV_IMPROVED)\s*\]", agent_text, re.IGNORECASE)
- if match:
- st = match.group(1).upper()
- if st == "DEV_IMPROVED":
- return "PASS", agent_text
- return st, agent_text
- return "ISSUES", agent_text
+try:
+ from worker.execution import ExecutionLimits, evaluate_agent_status, limits_for_task, run_bounded_shell, write_diagnostic
+except ImportError:
+ from skills.swarm.scripts.worker.execution import ExecutionLimits, evaluate_agent_status, limits_for_task, run_bounded_shell, write_diagnostic
try:
@@ -811,11 +796,15 @@ def execute_task(
task_index: int = 0,
messenger: Optional[Any] = None,
orchestrator_timeout: float = 600.0,
+ *,
+ limits: Optional[ExecutionLimits] = None,
) -> Tuple[str, str]:
task_type = "agent"
if isinstance(task_item, dict):
task_type = task_item.get("type", "agent")
+ limits = limits or limits_for_task(task_item, command_default=900 if task_type == "command" else 300)
+
# Command execution runner
if task_type == "command":
cmd_str = ""
@@ -828,26 +817,9 @@ def execute_task(
emit_tool_invocation(task_index, "run_command", {"command": cmd_str})
emit_milestone(task_index, "RUNNING_COMMAND", cmd_str)
print(f"Executing command task: {cmd_str}", flush=True)
- timeout = int(os.environ.get("COMMAND_TIMEOUT", "900"))
- try:
- res = subprocess.run(
- cmd_str,
- shell=True,
- cwd=repo_dir,
- capture_output=True,
- text=True,
- timeout=timeout,
- )
- if res.returncode == 0:
- emit_tool_result(task_index, "run_command", res.stdout, success=True)
- return "PASS", res.stdout
- else:
- err_msg = res.stderr if res.stderr else res.stdout
- emit_tool_result(task_index, "run_command", err_msg, success=False)
- return "ISSUES", err_msg
- except subprocess.TimeoutExpired:
- emit_tool_result(task_index, "run_command", f"Command timed out after {timeout} seconds", success=False)
- return "ISSUES", f"Command timed out after {timeout} seconds: {cmd_str}"
+ res, output = run_bounded_shell(cmd_str, repo_dir, limits.command_timeout_s)
+ emit_tool_result(task_index, "run_command", output, success=res.returncode == 0)
+ return ("PASS" if res.returncode == 0 else "ISSUES"), output
resolved_model = resolve_worker_model(raw_model=model_override, task_item=task_item)
if is_deepseek_model(resolved_model):
@@ -862,6 +834,7 @@ def execute_task(
project=project,
location=location_override,
messenger=messenger,
+ limits=limits,
)
if is_partner_model(resolved_model):
@@ -875,6 +848,7 @@ def execute_task(
project=project,
location=location,
messenger=messenger,
+ limits=limits,
)
# General Agent Execution using Google Antigravity SDK
@@ -1031,6 +1005,8 @@ def main() -> None:
)
sys.exit(1)
+ limits = limits_for_task(task_item, command_default=900 if isinstance(task_item, dict) and task_item.get("type") == "command" else 300)
+
task_model = resolve_worker_model(raw_model=model_override, task_item=task_item)
use_vertex = (
os.environ.get("USE_VERTEX_AI", "").lower() in ("1", "true", "yes")
@@ -1144,22 +1120,11 @@ def main() -> None:
task_index=task_index,
messenger=messenger,
orchestrator_timeout=orchestrator_timeout,
+ limits=limits,
)
- if agent_status == "BLOCKED":
- emit_milestone(task_index, "COMPLETE", "BLOCKED")
- print("=" * 80)
- print("[STATUS: BLOCKED]")
- print("Evidence:")
- print(f"- Task Index: {task_index}")
- print(f"- Branch: {branch_name}")
- print("- Commit SHA: N/A")
- print("- Changes Pushed: False")
- print("- Diff Stat:\nExecution blocked before changes.")
- print("Summary:")
- print(agent_summary.strip())
- print("=" * 80, flush=True)
- sys.exit(1)
+ agent_summary = scrub_credentials(agent_summary)
+ write_diagnostic(repo_dir, "agent-result", agent_summary)
score, accuracy, latency_ms = parse_score_metrics(repo_dir)
gcs_bucket = os.environ.get("GCS_RESULTS_BUCKET", "").strip()
@@ -1209,53 +1174,38 @@ def main() -> None:
agent_summary += f"\n[Steering Instructions Applied]: {steer_summary}"
print(f"Notice: Applied {len(steer_messages)} steering instruction(s): {steer_summary}", flush=True)
- # Prevent ghost commits: Only commit when agent_status == "PASS" and candidate files exist
- if agent_status == "PASS" and candidate_files:
+ if agent_status == "PASS" and verify_cmd_str:
emit_milestone(task_index, "VALIDATING_CANDIDATE")
- if verify_cmd_str:
- print(f"Running verification command: {verify_cmd_str}", flush=True)
- verify_timeout = int(os.environ.get("VERIFY_TIMEOUT", "300"))
- try:
- test_check = subprocess.run(
- verify_cmd_str,
- shell=True,
- cwd=repo_dir,
- capture_output=True,
- text=True,
- timeout=verify_timeout,
- )
- if test_check.returncode != 0:
- agent_status = "ISSUES"
- err_snippet = (test_check.stderr.strip() or test_check.stdout.strip())[-500:]
- diff_stat = f"Commit rejected: Candidate failed verification command (exit code {test_check.returncode}): {err_snippet}"
- print(f"Warning: Candidate modifications failed verification command (exit code {test_check.returncode}). Commit blocked.", file=sys.stderr)
- except subprocess.TimeoutExpired:
- agent_status = "ISSUES"
- diff_stat = f"Commit rejected: Candidate verification command timed out after {verify_timeout}s."
- print(f"Warning: Candidate verification command timed out after {verify_timeout}s. Commit blocked.", file=sys.stderr)
-
- if agent_status == "PASS":
- try:
- for cand_file in candidate_files:
- run_command(["git", "add", "--", cand_file], cwd=repo_dir, check=True)
- cached_diff = run_command(["git", "diff", "--cached", "--name-only"], cwd=repo_dir, check=False)
- if not cached_diff.stdout.strip():
- diff_stat = "Commit skipped: No candidate changes staged."
- print("Notice: No candidate changes staged; skipping git commit.", file=sys.stderr)
- else:
- commit_msg = f"worker-{task_index}: execute swarm brief"
- run_command(["git", "commit", "-m", commit_msg], cwd=repo_dir, check=True)
- commit_res = run_command(["git", "rev-parse", "HEAD"], cwd=repo_dir, check=True)
- commit_sha = commit_res.stdout.strip()
- diff_res = run_command(["git", "diff", "--stat", "HEAD~1", "HEAD"], cwd=repo_dir, check=False)
- diff_stat = diff_res.stdout.strip()
- except subprocess.CalledProcessError as exc:
- agent_status = "ISSUES"
- commit_sha = None
- err_text = (exc.stderr or str(exc)).strip()
- diff_stat = f"Commit failed: {err_text}"
- err_msg = exc.stderr.replace(gh_token, "REDACTED") if gh_token and gh_token in exc.stderr else err_text
- print(f"Warning: Git commit preparation failed: {err_msg}", file=sys.stderr)
+ test_check, verify_output = run_bounded_shell(
+ verify_cmd_str, repo_dir, limits.verify_timeout_s, label="verification",
+ )
+ agent_summary += f"\nVerification:\n{verify_output}"
+ if test_check.returncode != 0:
+ agent_status = "ISSUES"
+ diff_stat = f"Commit rejected: Candidate failed verification command (exit code {test_check.returncode})."
+
+ if agent_status == "PASS" and candidate_files:
+ try:
+ for cand_file in candidate_files:
+ run_command(["git", "add", "--", cand_file], cwd=repo_dir, check=True)
+ cached_diff = run_command(["git", "diff", "--cached", "--name-only"], cwd=repo_dir, check=False)
+ if not cached_diff.stdout.strip():
+ diff_stat = "Commit skipped: No candidate changes staged."
+ print("Notice: No candidate changes staged; skipping git commit.", file=sys.stderr)
+ else:
+ commit_msg = f"worker-{task_index}: execute swarm brief"
+ run_command(["git", "commit", "-m", commit_msg], cwd=repo_dir, check=True)
+ commit_res = run_command(["git", "rev-parse", "HEAD"], cwd=repo_dir, check=True)
+ commit_sha = commit_res.stdout.strip()
+ diff_res = run_command(["git", "diff", "--stat", "HEAD~1", "HEAD"], cwd=repo_dir, check=False)
+ diff_stat = diff_res.stdout.strip()
+ except subprocess.CalledProcessError as exc:
+ agent_status = "ISSUES"
+ commit_sha = None
+ err_text = (exc.stderr or str(exc)).strip()
+ diff_stat = f"Commit failed: {err_text}"
+ err_msg = exc.stderr.replace(gh_token, "REDACTED") if gh_token and gh_token in exc.stderr else err_text
+ print(f"Warning: Git commit preparation failed: {err_msg}", file=sys.stderr)
elif agent_status != "PASS":
diff_stat = f"Commit skipped: worker status is {agent_status} (PASS required; ghost commits prohibited)."
print(f"Notice: Agent status is {agent_status}. Skipping git commit.", file=sys.stderr)
@@ -1303,6 +1253,7 @@ def main() -> None:
status=final_status,
candidate_files=candidate_files,
commit_sha=commit_sha,
+ evidence_files=task_item.get("evidence_files", []) if isinstance(task_item, dict) else [],
)
if gcs_uris:
emit_milestone(task_index, "ARTIFACTS_UPLOADED")
@@ -1320,11 +1271,8 @@ def main() -> None:
except Exception:
pass
- # Fail-closed check: if final_status == "PASS" and candidate_files were modified,
- # but upload_run_artifacts() failed to upload patch.diff or status.json to GCS,
- # or if upload explicitly failed, override final_status = "ISSUES".
- if final_status == "PASS" and candidate_files:
- if upload_failed or "patch.diff" not in gcs_uris or "status.json" not in gcs_uris:
+ if final_status == "PASS":
+ if upload_failed or "status.json" not in gcs_uris or (candidate_files and "patch.diff" not in gcs_uris):
final_status = "ISSUES"
fail_evidence = "Artifact upload to GCS failed; candidate patch could not be preserved."
agent_summary += f"\n{fail_evidence}"
@@ -1339,7 +1287,6 @@ def main() -> None:
emit_milestone(task_index, "COMPLETE", final_status)
print("=" * 80)
- print(f"[STATUS: {final_status}]")
print("Evidence:")
print(f"- Task Index: {task_index}")
print(f"- Branch: {branch_name}")
@@ -1368,8 +1315,12 @@ def main() -> None:
print(f"- GCS Score: {gcs_uris['score.json']}")
print(f"- Diff Stat:\n{diff_stat}")
print("Summary:")
- print(agent_summary.strip())
- print("=" * 80, flush=True)
+ for summary_line in agent_summary.strip().splitlines():
+ print(f"> {summary_line}")
+ print("=" * 80)
+ print(f"[STATUS: {final_status}]", flush=True)
+ if final_status != "PASS":
+ sys.exit(1)
if __name__ == "__main__":
diff --git a/skills/swarm/scripts/result_harvester.py b/skills/swarm/scripts/result_harvester.py
index 67aac79..48eb40a 100644
--- a/skills/swarm/scripts/result_harvester.py
+++ b/skills/swarm/scripts/result_harvester.py
@@ -78,14 +78,17 @@ def harvest_gcs_results(
target_filename: str = "score.json",
) -> List[Dict[str, Any]]:
clean_prefix = prefix.strip("/")
+ storage_prefix = f"{clean_prefix}/" if clean_prefix else ""
try:
from google.cloud import storage
client = storage.Client()
bucket = client.bucket(bucket_name)
- blobs = bucket.list_blobs(prefix=clean_prefix)
+ blobs = bucket.list_blobs(prefix=storage_prefix)
results: List[Dict[str, Any]] = []
for blob in blobs:
+ if not blob.name.startswith(storage_prefix):
+ continue
if blob.name.endswith(target_filename):
payload = json.loads(blob.download_as_text(encoding="utf-8"))
results.append(payload)
@@ -126,6 +129,7 @@ def harvest_candidate_patches(
dest = Path(dest_dir)
dest.mkdir(parents=True, exist_ok=True)
clean_prefix = prefix.strip("/")
+ storage_prefix = f"{clean_prefix}/" if clean_prefix else ""
tasks_found: Dict[int, Dict[str, Any]] = {}
local_dir = os.environ.get("STORAGE_MESSENGER_LOCAL_DIR")
@@ -135,7 +139,7 @@ def harvest_candidate_patches(
for task_folder in sorted(src_base.iterdir()):
if not task_folder.is_dir():
continue
- m = re.search(r"task-(\d+)", task_folder.name)
+ m = re.fullmatch(r"task-(\d+)", task_folder.name)
if not m:
continue
task_idx = int(m.group(1))
@@ -159,6 +163,12 @@ def harvest_candidate_patches(
target_status = target_task_dir / "status.json"
shutil.copy2(status_src, target_status)
stf = target_status
+ for filename in ("partial.patch.diff", "results.tsv"):
+ source = task_folder / filename
+ if source.is_file():
+ shutil.copy2(source, target_task_dir / filename)
+ if (task_folder / "evidence").is_dir():
+ shutil.copytree(task_folder / "evidence", target_task_dir / "evidence", dirs_exist_ok=True)
tasks_found[task_idx] = {"score_file": sf, "patch_file": pf, "status_file": stf}
if not tasks_found:
@@ -167,20 +177,27 @@ def harvest_candidate_patches(
client = storage.Client()
bucket = client.bucket(bucket_name)
- blobs = bucket.list_blobs(prefix=clean_prefix)
+ blobs = bucket.list_blobs(prefix=storage_prefix)
for blob in blobs:
- m = re.search(r"task-(\d+)/([^/]+)$", blob.name)
+ if not blob.name.startswith(storage_prefix):
+ continue
+ rel_name = blob.name[len(storage_prefix):]
+ m = re.match(r"^task-(\d+)/(.+)$", rel_name)
if not m:
continue
task_idx = int(m.group(1))
filename = m.group(2)
- if filename not in ("score.json", "patch.diff", "status.json"):
+ relative = Path(filename)
+ if relative.is_absolute() or ".." in relative.parts:
+ continue
+ if filename not in ("score.json", "patch.diff", "partial.patch.diff", "status.json", "results.tsv") and not filename.startswith("evidence/"):
continue
if task_idx not in tasks_found:
tasks_found[task_idx] = {}
target_task_dir = dest / f"task-{task_idx}"
target_task_dir.mkdir(parents=True, exist_ok=True)
target_file = target_task_dir / filename
+ target_file.parent.mkdir(parents=True, exist_ok=True)
blob.download_to_filename(str(target_file))
if filename == "score.json":
tasks_found[task_idx]["score_file"] = target_file
@@ -201,7 +218,7 @@ def harvest_candidate_patches(
)
for item in dest.glob("task-*"):
if item.is_dir():
- m = re.search(r"task-(\d+)", item.name)
+ m = re.fullmatch(r"task-(\d+)", item.name)
if m:
task_idx = int(m.group(1))
sf = item / "score.json"
@@ -230,7 +247,6 @@ def harvest_candidate_patches(
try:
data = json.loads(Path(score_file).read_text(encoding="utf-8"))
score = data.get("score")
- status = data.get("status", "PASS" if score is not None else "UNKNOWN")
except Exception:
pass
@@ -250,8 +266,13 @@ def harvest_candidate_patches(
if isinstance(score, (int, float)) and isinstance(baseline_score, (int, float)):
score_delta = round(float(score) - float(baseline_score), 4)
+ if status != "PASS":
+ patch_file = None
+ partial_patch = dest / f"task-{task_idx}" / "partial.patch.diff"
results.append({
"task_index": task_idx,
+ "partial_patch_file": str(partial_patch) if partial_patch.is_file() else None,
+ "evidence_dir": str(dest / f"task-{task_idx}" / "evidence"),
"score": score,
"score_delta": score_delta,
"status": status,
diff --git a/skills/swarm/scripts/storage_uploader.py b/skills/swarm/scripts/storage_uploader.py
index 411405e..aa766c2 100644
--- a/skills/swarm/scripts/storage_uploader.py
+++ b/skills/swarm/scripts/storage_uploader.py
@@ -1,4 +1,5 @@
#!/usr/bin/env python3
+import fnmatch
import json
import os
import subprocess
@@ -9,6 +10,71 @@
from pathlib import Path
from typing import Dict, List, Optional, Union
+try:
+ from worker.execution import CREDENTIAL_PATTERNS
+except ImportError:
+ from skills.swarm.scripts.worker.execution import CREDENTIAL_PATTERNS
+
+
+def is_sensitive_path(path: Path, root: Path) -> bool:
+ try:
+ p = Path(path)
+ root_path = Path(root)
+ full_path = p if p.is_absolute() else (root_path / p)
+ if os.path.islink(full_path) or full_path.is_symlink():
+ return True
+ r = root_path.resolve()
+ resolved = full_path.resolve()
+ curr = full_path.parent
+ while curr != root_path and curr.resolve() != r and curr != curr.parent:
+ if os.path.islink(curr) or curr.is_symlink():
+ return True
+ curr = curr.parent
+ if r != resolved and r not in resolved.parents:
+ return True
+ parts = resolved.relative_to(r).parts
+ if ".git" in parts or ".git" in p.parts or ".git" in full_path.parts:
+ return True
+ sensitive_patterns = [
+ ".env",
+ ".env.*",
+ "*.pem",
+ "*.key",
+ "id_rsa*",
+ "id_ed25519*",
+ "credentials.json",
+ "token.json",
+ ]
+ for pattern in sensitive_patterns:
+ if fnmatch.fnmatch(full_path.name.lower(), pattern.lower()):
+ return True
+ if fnmatch.fnmatch(resolved.name.lower(), pattern.lower()):
+ return True
+ for part in parts:
+ for pattern in sensitive_patterns:
+ if fnmatch.fnmatch(part.lower(), pattern.lower()):
+ return True
+ return False
+ except Exception:
+ return True
+
+
+def file_contains_credentials(path: Path) -> bool:
+ p = Path(path)
+ if not p.is_file():
+ return False
+ try:
+ content = p.read_text(encoding="utf-8", errors="ignore")
+ except Exception:
+ try:
+ content = p.read_bytes().decode("utf-8", errors="ignore")
+ except Exception:
+ return False
+ for pat in CREDENTIAL_PATTERNS:
+ if pat.search(content):
+ return True
+ return False
+
def get_oauth_token() -> Optional[str]:
env_token = os.environ.get("GCS_OAUTH_TOKEN") or os.environ.get("GOOGLE_OAUTH_ACCESS_TOKEN")
@@ -33,9 +99,7 @@ def get_oauth_token() -> Optional[str]:
import google.auth
import google.auth.transport.requests
- credentials, _ = google.auth.default(
- scopes=["https://www.googleapis.com/auth/devstorage.read_write"]
- )
+ credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/devstorage.read_write"])
credentials.refresh(google.auth.transport.requests.Request())
if credentials.token:
return credentials.token
@@ -69,6 +133,7 @@ def upload_to_gcs(bucket_name: str, object_name: str, local_file_path: Path) ->
dest = Path(local_dir) / bucket_name / object_name
dest.parent.mkdir(parents=True, exist_ok=True)
import shutil
+
shutil.copy2(path, dest)
return True
@@ -256,50 +321,103 @@ def upload_run_artifacts(
status: str = "UNKNOWN",
candidate_files: Optional[List[str]] = None,
commit_sha: Optional[str] = None,
+ *,
+ evidence_files: Optional[List[str]] = None,
) -> Dict[str, str]:
repo = Path(repo_dir)
clean_prefix = prefix.strip().strip("/")
- task_dir = f"task-{task_index}"
- target_prefix = f"{clean_prefix}/{task_dir}" if clean_prefix else task_dir
+ target_prefix = f"{clean_prefix}/task-{task_index}" if clean_prefix else f"task-{task_index}"
uploaded: Dict[str, str] = {}
+ delivery_errors: List[str] = []
+ root = repo.resolve()
+
+ def _upload_checked(path: Path, name: str) -> bool:
+ if is_sensitive_path(path, root):
+ delivery_errors.append(f"{name} is sensitive or outside repository")
+ return False
+ if file_contains_credentials(path):
+ delivery_errors.append(f"{name} contains credentials")
+ return False
+ dest = f"{target_prefix}/{name}"
+ if upload_to_gcs(bucket_name, dest, path):
+ uploaded[name] = f"gs://{bucket_name}/{dest}"
+ return True
+ delivery_errors.append(f"{name} upload failed")
+ return False
+
+ evidence_paths: List[Path] = []
+ for relative in evidence_files or []:
+ raw_path = root / relative
+ if is_sensitive_path(raw_path, root):
+ delivery_errors.append(f"Evidence path is sensitive or outside repository: {relative}")
+ continue
+ if raw_path.is_dir():
+ for child in sorted(raw_path.rglob("*")):
+ if is_sensitive_path(child, root):
+ rel = child.relative_to(root).as_posix() if root in child.resolve().parents else child.name
+ delivery_errors.append(f"Evidence path is sensitive or outside repository: {rel}")
+ elif child.is_file():
+ evidence_paths.append(child)
+ elif raw_path.is_file():
+ evidence_paths.append(raw_path)
+ else:
+ delivery_errors.append(f"Evidence file missing: {relative}")
+
+ diagnostics = root / ".agystack/diagnostics"
+ if diagnostics.is_dir():
+ if is_sensitive_path(diagnostics, root):
+ delivery_errors.append(f"Diagnostic path is sensitive or outside repository: {diagnostics.name}")
+ else:
+ for diag in sorted(diagnostics.glob("*.log")):
+ if is_sensitive_path(diag, root):
+ delivery_errors.append(f"Diagnostic path is sensitive or outside repository: {diag.name}")
+ elif diag.is_file():
+ evidence_paths.append(diag)
+
+ for path in sorted(set(evidence_paths)):
+ resolved = path.resolve()
+ if not resolved.is_file():
+ continue
+ rel_name = path.relative_to(root).as_posix() if root in resolved.parents else path.name
+ _upload_checked(path, f"evidence/{rel_name}")
+
+ if delivery_errors and status == "PASS":
+ status = "ISSUES"
- # ATOMIC PUBLICATION:
- # Under zero-push architecture, if status == "PASS":
- # Generate patch.diff and upload it to GCS FIRST (target_prefix/patch.diff).
patch_uploaded = False
- if status == "PASS":
- patch_path = repo / "patch.diff"
- if not patch_path.is_file():
- patch_content = _generate_git_patch(repo, candidate_files=candidate_files, commit_sha=commit_sha)
- if patch_content:
- patch_path.write_text(patch_content, encoding="utf-8")
-
- if patch_path.is_file() and patch_path.stat().st_size > 0:
- dest = f"{target_prefix}/patch.diff"
- if upload_to_gcs(bucket_name, dest, patch_path):
- uploaded["patch.diff"] = f"gs://{bucket_name}/{dest}"
- patch_uploaded = True
-
- # If candidate_files were modified and patch.diff failed to upload or is missing/empty:
- # Override status = "ISSUES"!
+ if candidate_files or (status == "PASS" and candidate_files is None):
+ patch_name = "patch.diff" if status == "PASS" else "partial.patch.diff"
+ patch_path = repo / patch_name
+ if is_sensitive_path(patch_path, root):
+ delivery_errors.append(f"Candidate patch path is sensitive: {patch_name}")
+ if status == "PASS":
+ status = "ISSUES"
+ else:
+ if candidate_files or not patch_path.is_file():
+ patch_content = _generate_git_patch(repo, candidate_files=candidate_files, commit_sha=commit_sha)
+ if patch_content:
+ patch_path.write_text(patch_content, encoding="utf-8")
+
+ if patch_path.is_file() and patch_path.stat().st_size > 0:
+ patch_uploaded = _upload_checked(patch_path, patch_name)
+
if candidate_files and not patch_uploaded:
- status = "ISSUES"
+ delivery_errors.append("Candidate patch failed to upload or is missing under zero-push delivery.")
+ if status == "PASS":
+ status = "ISSUES"
- # Upload score.json and results.tsv
score_path = repo / "score.json"
if score_path.is_file():
- dest = f"{target_prefix}/score.json"
- if upload_to_gcs(bucket_name, dest, score_path):
- uploaded["score.json"] = f"gs://{bucket_name}/{dest}"
+ _upload_checked(score_path, "score.json")
results_path = repo / "results.tsv"
if results_path.is_file():
- dest = f"{target_prefix}/results.tsv"
- if upload_to_gcs(bucket_name, dest, results_path):
- uploaded["results.tsv"] = f"gs://{bucket_name}/{dest}"
+ _upload_checked(results_path, "results.tsv")
+
+ if delivery_errors and status == "PASS":
+ status = "ISSUES"
- # Write and upload status.json LAST with the finalized status.
status_path = repo / "status.json"
status_data = {}
if status_path.is_file():
@@ -309,12 +427,18 @@ def upload_run_artifacts(
status_data = {}
status_data["task_index"] = task_index
status_data["status"] = status
- if candidate_files and not patch_uploaded and status == "ISSUES":
- status_data["evidence"] = "Candidate patch failed to upload or is missing under zero-push delivery."
+ status_data["artifacts"] = uploaded
+ if delivery_errors:
+ status_data["evidence"] = "\n".join(delivery_errors)
+ status_data["delivery_errors"] = delivery_errors
+ else:
+ status_data.pop("delivery_errors", None)
status_path.write_text(json.dumps(status_data, indent=2), encoding="utf-8")
dest = f"{target_prefix}/status.json"
if upload_to_gcs(bucket_name, dest, status_path):
uploaded["status.json"] = f"gs://{bucket_name}/{dest}"
+ else:
+ delivery_errors.append("status.json upload failed")
return uploaded
diff --git a/skills/swarm/scripts/worker/execution.py b/skills/swarm/scripts/worker/execution.py
new file mode 100644
index 0000000..6ae710b
--- /dev/null
+++ b/skills/swarm/scripts/worker/execution.py
@@ -0,0 +1,183 @@
+import os
+import re
+import signal
+import subprocess
+import threading
+import time
+import uuid
+from collections.abc import Mapping
+from dataclasses import dataclass, fields
+from pathlib import Path
+
+LIMIT_ENV = {
+ "max_agent_turns": "AGYSTACK_MAX_AGENT_TURNS",
+ "command_timeout_s": "COMMAND_TIMEOUT",
+ "verify_timeout_s": "VERIFY_TIMEOUT",
+}
+TERMINAL_STATES = frozenset({"PASS", "ISSUES", "BLOCKED"})
+_UNSET = object()
+
+
+def _positive_int(value, name):
+ if isinstance(value, str) and re.fullmatch(r"[0-9]+", value):
+ value = int(value)
+ if type(value) is not int or value <= 0:
+ raise ValueError(f"{name} must be a positive integer")
+ return value
+
+
+@dataclass(frozen=True)
+class ExecutionLimits:
+ max_agent_turns: int = 30
+ command_timeout_s: int = 300
+ verify_timeout_s: int = 300
+
+ def __post_init__(self):
+ for field in fields(self):
+ if type(getattr(self, field.name)) is not int:
+ raise ValueError(f"{field.name} must be a positive integer")
+ _positive_int(getattr(self, field.name), field.name)
+
+
+def _limit_fields(value):
+ if not isinstance(value, Mapping):
+ raise ValueError("execution_limits must be an object")
+ unknown = value.keys() - LIMIT_ENV.keys()
+ if unknown:
+ raise ValueError(f"Unknown execution_limits fields: {sorted(unknown)}")
+ for name, raw in value.items():
+ if type(raw) is not int:
+ raise ValueError(f"{name} must be a positive integer")
+ return {name: _positive_int(raw, name) for name, raw in value.items()}
+
+
+def resolve_execution_limits(task_limits=_UNSET, *, environ=None, command_default=300):
+ env = os.environ if environ is None else environ
+ values = dict(max_agent_turns=30, command_timeout_s=command_default, verify_timeout_s=300)
+ values.update({name: _positive_int(env[key], key) for name, key in LIMIT_ENV.items() if key in env})
+ if task_limits is not _UNSET:
+ values.update(_limit_fields(task_limits))
+ return ExecutionLimits(**values)
+
+
+def limits_for_task(task_item, *, command_default=300):
+ overrides = task_item.get("execution_limits", {}) if isinstance(task_item, dict) else {}
+ return resolve_execution_limits(overrides, command_default=command_default)
+
+
+def evaluate_agent_status(agent_text):
+ declarations = []
+ lines = agent_text.rstrip().splitlines()
+ fence = None
+ for index, line in enumerate(lines):
+ stripped = line.strip()
+ marker = re.match(r"^(`{3,}|~{3,})", stripped)
+ if marker:
+ token = marker.group(1)
+ if fence is None:
+ fence = token
+ elif token[0] == fence[0] and len(token) >= len(fence) and stripped == token:
+ fence = None
+ continue
+ if fence is not None or line.startswith((" ", "\t")):
+ continue
+ match = re.fullmatch(r"\[STATUS: (PASS|ISSUES|BLOCKED)\]", stripped)
+ if match:
+ declarations.append((index, match.group(1)))
+ elif re.fullmatch(r"\[STATUS\b.*\]", stripped, re.IGNORECASE):
+ declarations.append((index, "ISSUES"))
+ if len(declarations) == 1 and declarations[0][0] == len(lines) - 1:
+ return declarations[0][1], agent_text
+ return "ISSUES", agent_text
+
+
+CREDENTIAL_PATTERNS = [
+ re.compile(r"ghp_[A-Za-z0-9_]{36,}"),
+ re.compile(r"AIza[0-9A-Za-z-_]{35}"),
+ re.compile(r"github_pat_[0-9A-Za-z_]{20,}"),
+ re.compile(r"ya29\.[0-9A-Za-z-_]+"),
+ re.compile(r"Bearer [A-Za-z0-9_\-\.]+"),
+]
+
+
+def scrub_credentials(text):
+ for pattern in CREDENTIAL_PATTERNS:
+ text = pattern.sub("[REDACTED]", text)
+ return text
+
+
+def write_diagnostic(repo_dir, name, text):
+ directory = Path(repo_dir) / ".agystack/diagnostics"
+ directory.mkdir(parents=True, exist_ok=True)
+ path = directory / f"{name}-{uuid.uuid4().hex}.log"
+ path.write_text(scrub_credentials(text)[-262144:], encoding="utf-8")
+ return path
+
+
+def exhaustion_result(repo_dir, limit, consumed, last_tool, last_response):
+ summary = (
+ f"Agent turn limit exhausted: limit={limit}, consumed={consumed}, "
+ f"last_tool={last_tool or 'none'}, last_response={last_response[-4096:]!r}"
+ )
+ summary = scrub_credentials(summary)
+ write_diagnostic(repo_dir, "agent-exhaustion", summary)
+ return "ISSUES", summary
+
+
+def run_bounded_shell(command, repo_dir, timeout_s, *, label="command"):
+ buffers = [bytearray(), bytearray()]
+ shell_args = ["bash", "-o", "pipefail", "-c", command] if os.name == "posix" else command
+ try:
+ process = subprocess.Popen(
+ shell_args,
+ shell=os.name != "posix",
+ cwd=repo_dir,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ start_new_session=os.name == "posix",
+ )
+ except OSError as exc:
+ error = scrub_credentials(str(exc))
+ log_path = write_diagnostic(repo_dir, label, f"Command could not start: {error}")
+ return subprocess.CompletedProcess(command, 127, "", error), f"Exit code 127: {error}\nLog: {log_path}"
+
+ def drain(stream, buffer):
+ with stream:
+ while chunk := os.read(stream.fileno(), 8192):
+ buffer.extend(chunk)
+ del buffer[:-131072]
+
+ readers = [
+ threading.Thread(target=drain, args=(stream, buffer), daemon=True)
+ for stream, buffer in zip((process.stdout, process.stderr), buffers)
+ ]
+ for reader in readers:
+ reader.start()
+ timed_out = False
+ deadline = time.monotonic() + timeout_s
+ try:
+ process.wait(timeout=timeout_s)
+ for reader in readers:
+ reader.join(timeout=max(0, deadline - time.monotonic()))
+ if any(reader.is_alive() for reader in readers):
+ raise subprocess.TimeoutExpired(command, timeout_s)
+ except subprocess.TimeoutExpired:
+ timed_out = True
+ if os.name == "posix":
+ try:
+ os.killpg(process.pid, signal.SIGKILL)
+ except ProcessLookupError:
+ pass
+ else:
+ process.kill()
+ process.wait()
+ for reader in readers:
+ reader.join(timeout=1)
+ stdout, stderr = [scrub_credentials(bytes(buffer).decode("utf-8", errors="replace")) for buffer in buffers]
+ if timed_out:
+ stderr += f"\nCommand timed out after {timeout_s} seconds."
+ returncode = 124 if timed_out else process.returncode
+ output = f"Exit code {returncode}:\n{stdout}\n{stderr}"
+ log_path = write_diagnostic(repo_dir, label, output)
+ excerpt = f"Exit code {returncode}:\n{stdout[-8192:]}\n{stderr[-8192:]}\nLog: {log_path}"
+ return subprocess.CompletedProcess(command, returncode, stdout, stderr), excerpt
diff --git a/skills/swarm/scripts/worker/partner_engine.py b/skills/swarm/scripts/worker/partner_engine.py
index b1cedd5..35fa420 100644
--- a/skills/swarm/scripts/worker/partner_engine.py
+++ b/skills/swarm/scripts/worker/partner_engine.py
@@ -6,12 +6,13 @@
"""
import os
-import subprocess
import sys
import time
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
+from .execution import ExecutionLimits, evaluate_agent_status as _evaluate_agent_status, exhaustion_result, limits_for_task, run_bounded_shell
+
try:
from pre_tool_safety import evaluate_safety
except ImportError:
@@ -30,7 +31,7 @@
1. Implement the requested task directly and surgically in the workspace.
2. Maintain high engineering rigor: run test suites and verification commands before completing.
3. Keep diffs minimal and clean. Avoid narrating comments or speculative code.
-4. Conclude with a clear summary and status: you MUST include either [STATUS: PASS] or [STATUS: ISSUES] at the end of your final response.
+4. End your final response with exactly one standalone unquoted line, outside code fences: [STATUS: PASS], [STATUS: ISSUES], or [STATUS: BLOCKED].
"""
@@ -288,7 +289,7 @@ def _safe_resolve_repo_path(repo_dir: Path, rel_path: str) -> Optional[Path]:
return None
-def execute_workspace_tool(tool_name: str, tool_input: Dict[str, Any], repo_dir: Path) -> str:
+def execute_workspace_tool(tool_name: str, tool_input: Dict[str, Any], repo_dir: Path, *, limits: Optional[ExecutionLimits] = None) -> str:
try:
if tool_name == "read_file":
path_str = str(tool_input.get("path", "")).strip()
@@ -352,27 +353,9 @@ def execute_workspace_tool(tool_name: str, tool_input: Dict[str, Any], repo_dir:
reason = safety_res.get("reason", "Destructive command rejected.")
return f"Error: Command blocked by safety policy ({safety_res.get('decision')}): {reason}"
- timeout = int(os.environ.get("COMMAND_TIMEOUT", "300"))
- try:
- res = subprocess.run(
- cmd_str,
- shell=True,
- cwd=repo_dir,
- capture_output=True,
- text=True,
- timeout=timeout,
- )
- output = res.stdout
- if res.stderr:
- if output:
- output += "\n" + res.stderr
- else:
- output = res.stderr
- return f"Exit code {res.returncode}:\n{output}"
- except subprocess.TimeoutExpired:
- return f"Error: command timed out after {timeout} seconds: {cmd_str}"
- except Exception as e:
- return f"Error executing command: {e}"
+ limits = limits or limits_for_task(None)
+ _, output = run_bounded_shell(cmd_str, repo_dir, limits.command_timeout_s)
+ return output
else:
return f"Error: unknown tool \x27{tool_name}\x27."
@@ -495,7 +478,10 @@ def execute_claude_model_agent(
project: Optional[str] = None,
location: Optional[str] = None,
messenger: Optional[Any] = None,
+ *,
+ limits: Optional[ExecutionLimits] = None,
) -> Tuple[str, str]:
+ limits = limits or limits_for_task(task_item)
anthropic = _get_anthropic()
if anthropic is None:
return "BLOCKED", "Missing dependency: anthropic[vertex] Python package is not installed."
@@ -527,10 +513,14 @@ def execute_claude_model_agent(
messages: List[Dict[str, Any]] = [
{"role": "user", "content": task_brief}
]
- max_turns = 30
+ max_turns = limits.max_agent_turns
+ consumed = 0
+ last_tool = ""
last_text = ""
for turn in range(max_turns):
+ last_text = ""
+ consumed += 1
response = client.messages.create(
model=model,
max_tokens=4096,
@@ -584,7 +574,8 @@ def execute_claude_model_agent(
tool_results = []
for tool_id, tool_name, tool_input in tool_uses:
emit_tool_invocation(task_index, tool_name, tool_input)
- result_str = execute_workspace_tool(tool_name, tool_input, repo_dir)
+ last_tool = tool_name
+ result_str = execute_workspace_tool(tool_name, tool_input, repo_dir, limits=limits)
emit_tool_result(task_index, tool_name, result_str)
tool_results.append({
"type": "tool_result",
@@ -597,7 +588,7 @@ def execute_claude_model_agent(
"content": tool_results,
})
- return _evaluate_agent_status(last_text)
+ return exhaustion_result(repo_dir, max_turns, consumed, last_tool, last_text)
except Exception as exc:
return "ISSUES", f"Partner model execution exception: {exc}"
@@ -611,7 +602,10 @@ def execute_fable_model_agent(
project: Optional[str] = None,
location: Optional[str] = None,
messenger: Optional[Any] = None,
+ *,
+ limits: Optional[ExecutionLimits] = None,
) -> Tuple[str, str]:
+ limits = limits or limits_for_task(task_item)
aiplatform = _get_aiplatform()
if aiplatform is None:
@@ -655,11 +649,15 @@ def execute_fable_model_agent(
messages: List[Dict[str, Any]] = [
{"role": "user", "content": task_brief}
]
- max_turns = 30
+ max_turns = limits.max_agent_turns
+ consumed = 0
+ last_tool = ""
last_text = ""
for turn in range(max_turns):
+ last_text = ""
response = None
+ consumed += 1
if endpoint is not None and hasattr(endpoint, "predict"):
response = endpoint.predict(
instances=[{
@@ -738,7 +736,8 @@ def execute_fable_model_agent(
tool_results = []
for tool_id, tool_name, tool_input in tool_uses:
emit_tool_invocation(task_index, tool_name, tool_input)
- result_str = execute_workspace_tool(tool_name, tool_input, repo_dir)
+ last_tool = tool_name
+ result_str = execute_workspace_tool(tool_name, tool_input, repo_dir, limits=limits)
emit_tool_result(task_index, tool_name, result_str)
tool_results.append({
"type": "tool_result",
@@ -751,7 +750,7 @@ def execute_fable_model_agent(
"content": tool_results,
})
- return _evaluate_agent_status(last_text)
+ return exhaustion_result(repo_dir, max_turns, consumed, last_tool, last_text)
except Exception as exc:
return "ISSUES", f"Vertex Model Garden Fable execution exception: {exc}"
@@ -765,7 +764,10 @@ def execute_deepseek_model_agent(
project: Optional[str] = None,
location: Optional[str] = None,
messenger: Optional[Any] = None,
+ *,
+ limits: Optional[ExecutionLimits] = None,
) -> Tuple[str, str]:
+ limits = limits or limits_for_task(task_item)
genai = _get_genai()
if genai is None:
return "BLOCKED", "Missing dependency: google-genai Python package is not installed."
@@ -821,13 +823,21 @@ def execute_deepseek_model_agent(
parts=[genai.types.Part.from_text(text=task_brief)],
)
]
- max_turns = 30
+ max_turns = limits.max_agent_turns
+ consumed = 0
+ last_tool = ""
last_text = ""
for turn in range(max_turns):
+ if consumed >= max_turns:
+ return exhaustion_result(repo_dir, max_turns, consumed, last_tool, last_text)
+ last_text = ""
response = None
for attempt in range(5):
+ if consumed >= max_turns:
+ return exhaustion_result(repo_dir, max_turns, consumed, last_tool, last_text)
try:
+ consumed += 1
response = client.models.generate_content(
model=model,
contents=contents,
@@ -843,10 +853,12 @@ def execute_deepseek_model_agent(
or "resource_exhausted" in err_str
or "deadline" in err_str
or "connection" in err_str
- ) and attempt < 4:
+ ) and attempt < 4 and consumed < max_turns:
backoff = (2 ** attempt) * 2
emit_milestone(task_index, "RETRY", f"Transient {call_exc}; retrying in {backoff}s (attempt {attempt + 1}/5)...")
time.sleep(backoff)
+ elif consumed >= max_turns:
+ return exhaustion_result(repo_dir, max_turns, consumed, last_tool, str(call_exc))
else:
raise
@@ -882,6 +894,12 @@ def execute_deepseek_model_agent(
if fc:
func_calls.append(fc)
+ try:
+ response_text = getattr(response, "text", "")
+ last_text = response_text if isinstance(response_text, str) else ""
+ except Exception:
+ last_text = ""
+
if not func_calls:
resp_text = None
try:
@@ -903,19 +921,21 @@ def execute_deepseek_model_agent(
]
last_text = "\n".join(text_pieces).strip()
- if not last_text and turn < max_turns - 1:
+ if not last_text and consumed >= max_turns:
+ return exhaustion_result(repo_dir, max_turns, consumed, last_tool, last_text)
+ if not last_text:
contents.append(genai.types.Content(
role="model",
parts=[genai.types.Part.from_text(text="")]
))
contents.append(genai.types.Content(
role="user",
- parts=[genai.types.Part.from_text(text="Please state your final summary and conclusion status: [STATUS: PASS] or [STATUS: ISSUES].")]
+ parts=[genai.types.Part.from_text(text="Give your final summary. End with exactly one standalone unquoted status line outside code fences, [STATUS: PASS], [STATUS: ISSUES], or [STATUS: BLOCKED].")]
))
continue
cleaned_text = re.sub(r".*?", "", last_text, flags=re.DOTALL).strip()
- return _evaluate_agent_status(cleaned_text or last_text)
+ return _evaluate_agent_status(cleaned_text)
# Filter out empty text parts so DeepSeek does not echo empty text on next turn
clean_parts = [
@@ -935,7 +955,8 @@ def execute_deepseek_model_agent(
except Exception:
tool_input = {}
emit_tool_invocation(task_index, tool_name, tool_input)
- result_str = execute_workspace_tool(tool_name, tool_input, repo_dir)
+ last_tool = tool_name
+ result_str = execute_workspace_tool(tool_name, tool_input, repo_dir, limits=limits)
emit_tool_result(task_index, tool_name, result_str)
response_parts.append(
genai.types.Part.from_function_response(
@@ -946,8 +967,7 @@ def execute_deepseek_model_agent(
contents.append(genai.types.Content(role="user", parts=response_parts))
- cleaned_text = re.sub(r".*?", "", last_text, flags=re.DOTALL).strip()
- return _evaluate_agent_status(cleaned_text or last_text)
+ return exhaustion_result(repo_dir, max_turns, consumed, last_tool, last_text)
except Exception as exc:
return "ISSUES", f"DeepSeek model execution exception: {exc}"
@@ -961,7 +981,10 @@ def execute_partner_model_agent(
project: Optional[str] = None,
location: Optional[str] = None,
messenger: Optional[Any] = None,
+ *,
+ limits: Optional[ExecutionLimits] = None,
) -> Tuple[str, str]:
+ limits = limits or limits_for_task(task_item)
if is_deepseek_model(model):
return execute_deepseek_model_agent(
model=model,
@@ -972,6 +995,7 @@ def execute_partner_model_agent(
project=project,
location=location,
messenger=messenger,
+ limits=limits,
)
if is_fable_model(model):
return execute_fable_model_agent(
@@ -983,6 +1007,7 @@ def execute_partner_model_agent(
project=project,
location=location,
messenger=messenger,
+ limits=limits,
)
return execute_claude_model_agent(
model=model,
@@ -993,15 +1018,5 @@ def execute_partner_model_agent(
project=project,
location=location,
messenger=messenger,
+ limits=limits,
)
-
-
-def _evaluate_agent_status(agent_text: str) -> Tuple[str, str]:
- import re
- match = re.search(r"\[STATUS:\s*(PASS|ISSUES|BLOCKED|DEV_IMPROVED)\s*\]", agent_text, re.IGNORECASE)
- if match:
- st = match.group(1).upper()
- if st == "DEV_IMPROVED":
- return "PASS", agent_text
- return st, agent_text
- return "ISSUES", agent_text
diff --git a/tests/test_architecture_remediation.py b/tests/test_architecture_remediation.py
index 8f38cbe..9a8eee9 100755
--- a/tests/test_architecture_remediation.py
+++ b/tests/test_architecture_remediation.py
@@ -86,22 +86,20 @@ def test_cli_argument_resolution_parallelism_and_gcs_prefix(self):
def test_strict_status_regex_parsing(self):
# 1. Exact valid tokens
- status, _ = cloud_worker.evaluate_agent_status("Finished work. [STATUS: PASS]")
+ status, _ = cloud_worker.evaluate_agent_status("Finished work.\n[STATUS: PASS]")
self.assertEqual(status, "PASS")
status, _ = cloud_worker.evaluate_agent_status("[STATUS: ISSUES]\nFound some edge cases")
self.assertEqual(status, "ISSUES")
- status, _ = cloud_worker.evaluate_agent_status("[STATUS: BLOCKED ] Out of quota")
+ status, _ = cloud_worker.evaluate_agent_status("Out of quota\n[STATUS: BLOCKED]")
self.assertEqual(status, "BLOCKED")
- # Case-insensitive
status, _ = cloud_worker.evaluate_agent_status("[status: pass]")
- self.assertEqual(status, "PASS")
+ self.assertEqual(status, "ISSUES")
- # Support DEV_IMPROVED mapped to PASS
status, _ = cloud_worker.evaluate_agent_status("[STATUS: DEV_IMPROVED]\nCandidate improved score")
- self.assertEqual(status, "PASS")
+ self.assertEqual(status, "ISSUES")
# 2. False positives without strict token must fall back to ISSUES
status, _ = cloud_worker.evaluate_agent_status("I hope this passes and does not fail password test.")
@@ -145,9 +143,11 @@ def test_harvest_candidate_patches(self):
task1_dir.mkdir(parents=True, exist_ok=True)
(task0_dir / "score.json").write_text(json.dumps({"score": 98.2, "status": "PASS"}), encoding="utf-8")
+ (task0_dir / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
(task0_dir / "patch.diff").write_text("--- a/foo\n+++ b/foo\n+pass", encoding="utf-8")
(task1_dir / "score.json").write_text(json.dumps({"score": 45.0, "status": "ISSUES"}), encoding="utf-8")
+ (task1_dir / "status.json").write_text(json.dumps({"status": "ISSUES"}), encoding="utf-8")
(task1_dir / "patch.diff").write_text("--- a/bar\n+++ b/bar\n+fail", encoding="utf-8")
dest_dir = self.tmp_path / "downloaded_patches"
@@ -390,8 +390,11 @@ def test_harvest_candidate_patches_score_delta_ranking(self):
t.mkdir(parents=True, exist_ok=True)
(t0 / "score.json").write_text(json.dumps({"score": 90.0, "status": "PASS"}), encoding="utf-8")
+ (t0 / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
(t1 / "score.json").write_text(json.dumps({"score": 95.0, "status": "PASS"}), encoding="utf-8")
+ (t1 / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
(t2 / "score.json").write_text(json.dumps({"score": 99.0, "status": "ISSUES"}), encoding="utf-8")
+ (t2 / "status.json").write_text(json.dumps({"status": "ISSUES"}), encoding="utf-8")
(t3 / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
dest_dir = self.tmp_path / "harvest_delta_out"
@@ -432,12 +435,14 @@ def test_harvest_session_cli_standalone(self):
task_dir = mock_storage / bucket / f"swarms/{session_id}" / "task-0"
task_dir.mkdir(parents=True, exist_ok=True)
(task_dir / "score.json").write_text(json.dumps({"score": 92.5, "status": "PASS"}), encoding="utf-8")
+ (task_dir / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
(task_dir / "patch.diff").write_text("--- a/f\n+++ b/f\n+ok", encoding="utf-8")
cmd = [
sys.executable,
str(dispatch_script),
"--harvest-session", session_id,
+ "--tasks", "1",
"--gcs-bucket", bucket,
"--baseline-score", "90.0",
]
@@ -543,12 +548,15 @@ def test_result_harvester_sorts_non_numeric_and_nan_metrics(self):
# Task 0: string score "N/A"
(t0 / "score.json").write_text(json.dumps({"score": "N/A", "status": "PASS"}), encoding="utf-8")
+ (t0 / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
# Task 1: valid float score 95.0
(t1 / "score.json").write_text(json.dumps({"score": 95.0, "status": "PASS"}), encoding="utf-8")
+ (t1 / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
# Task 2: None score
(t2 / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
# Task 3: valid float score 99.0
(t3 / "score.json").write_text(json.dumps({"score": 99.0, "status": "PASS"}), encoding="utf-8")
+ (t3 / "status.json").write_text(json.dumps({"status": "PASS"}), encoding="utf-8")
dest_dir = self.tmp_path / "harvest_nan_out"
with patch.dict(os.environ, {"STORAGE_MESSENGER_LOCAL_DIR": str(mock_storage)}):
@@ -685,7 +693,9 @@ def fake_execute(*args, **kwargs):
with patch("cloud_worker.execute_task", side_effect=fake_execute):
import io
with patch("sys.stdout", new_callable=io.StringIO) as mock_out:
- cloud_worker.main()
+ with self.assertRaises(SystemExit) as exit_result:
+ cloud_worker.main()
+ self.assertEqual(exit_result.exception.code, 1)
out = mock_out.getvalue()
self.assertIn("[STATUS: ISSUES]", out)
self.assertIn("Fail-closed: No GCS bucket configured", out)
diff --git a/tests/test_cloud_preflight_and_worker.py b/tests/test_cloud_preflight_and_worker.py
index 93d5df9..2e8fc70 100644
--- a/tests/test_cloud_preflight_and_worker.py
+++ b/tests/test_cloud_preflight_and_worker.py
@@ -412,7 +412,7 @@ def test_agent_mode_passes_poteto_system_prompt_to_config(self):
mock_agent_instance = MagicMock()
mock_resp = MagicMock()
- mock_resp.text = AsyncMock(return_value="[STATUS: PASS] completed successfully")
+ mock_resp.text = AsyncMock(return_value="completed successfully\n[STATUS: PASS]")
mock_agent_instance.chat = AsyncMock(return_value=mock_resp)
mock_agent_cm = MagicMock()
@@ -449,7 +449,7 @@ def test_agent_mode_env_var_overrides_poteto_system_prompt(self):
mock_agent_instance = MagicMock()
mock_resp = MagicMock()
- mock_resp.text = AsyncMock(return_value="[STATUS: PASS] done")
+ mock_resp.text = AsyncMock(return_value="done\n[STATUS: PASS]")
mock_agent_instance.chat = AsyncMock(return_value=mock_resp)
mock_agent_cm = MagicMock()
@@ -488,7 +488,7 @@ def fake_local_agent_config(**kwargs):
mock_agent_instance = MagicMock()
mock_resp = MagicMock()
- mock_resp.text = AsyncMock(return_value="[STATUS: PASS] fallback success")
+ mock_resp.text = AsyncMock(return_value="fallback success\n[STATUS: PASS]")
mock_agent_instance.chat = AsyncMock(return_value=mock_resp)
mock_agent_cm = MagicMock()
@@ -1817,18 +1817,17 @@ def post_tool_call(self, fn):
def test_command_task_timeout_handling(self):
with tempfile.TemporaryDirectory() as tmp_dir:
repo_dir = Path(tmp_dir)
- with patch.dict(os.environ, {"COMMAND_TIMEOUT": "5"}), patch(
- "subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="sleep 100", timeout=5)
- ):
+ with patch.dict(os.environ, {"COMMAND_TIMEOUT": "1"}):
status, output = execute_task(
- task_item={"type": "command", "command": "sleep 100"},
+ task_item={"type": "command", "command": "echo started; sleep 2"},
task_brief="Run long command",
repo_dir=repo_dir,
model_override="flash",
api_key="test-key",
)
self.assertEqual(status, "ISSUES")
- self.assertEqual(output, "Command timed out after 5 seconds: sleep 100")
+ self.assertIn("timed out after 1 seconds", output)
+ self.assertIn("started", output)
if __name__ == "__main__":
diff --git a/tests/test_cloud_worker_vertex_partner.py b/tests/test_cloud_worker_vertex_partner.py
index a9b641e..8a93910 100644
--- a/tests/test_cloud_worker_vertex_partner.py
+++ b/tests/test_cloud_worker_vertex_partner.py
@@ -138,7 +138,7 @@ def test_adc_project_and_region_pass_through(self):
mock_client = MagicMock()
mock_response = MagicMock()
mock_block = MagicMock()
- mock_block.text = "[STATUS: PASS] Work verified."
+ mock_block.text = "Work verified.\n[STATUS: PASS]"
mock_response.content = [mock_block]
mock_client.messages.create.return_value = mock_response
mock_anthropic.AnthropicVertex.return_value = mock_client
@@ -164,7 +164,7 @@ def test_adc_env_var_precedence_and_us_central1_fallback(self):
mock_client = MagicMock()
mock_response = MagicMock()
mock_block = MagicMock()
- mock_block.text = "[STATUS: PASS] Success."
+ mock_block.text = "Success.\n[STATUS: PASS]"
mock_response.content = [mock_block]
mock_client.messages.create.return_value = mock_response
mock_anthropic.AnthropicVertex.return_value = mock_client
@@ -224,7 +224,7 @@ def test_milestone_emission_partner_model(self):
mock_client = MagicMock()
mock_response = MagicMock()
mock_block = MagicMock()
- mock_block.text = "[STATUS: PASS] Done."
+ mock_block.text = "Done.\n[STATUS: PASS]"
mock_response.content = [mock_block]
mock_client.messages.create.return_value = mock_response
mock_anthropic.AnthropicVertex.return_value = mock_client
@@ -249,7 +249,7 @@ def test_milestone_emission_fable_model(self):
mock_aiplatform = MagicMock()
mock_endpoint = MagicMock()
mock_resp = MagicMock()
- mock_resp.predictions = ["[STATUS: PASS] Done."]
+ mock_resp.predictions = ["Done.\n[STATUS: PASS]"]
mock_endpoint.predict.return_value = mock_resp
mock_aiplatform.Endpoint.return_value = mock_endpoint
@@ -273,14 +273,14 @@ def test_milestone_emission_deepseek_model(self):
mock_genai = MagicMock()
mock_client = MagicMock()
mock_resp = MagicMock()
- mock_resp.text = "[STATUS: PASS] DeepSeek done."
+ mock_resp.text = "DeepSeek done.\n[STATUS: PASS]"
mock_resp.function_calls = []
mock_resp.candidates = [
MagicMock(
content=MagicMock(
parts=[
MagicMock(
- text="[STATUS: PASS] DeepSeek done.",
+ text="DeepSeek done.\n[STATUS: PASS]",
thought=None,
function_call=None,
)
@@ -336,7 +336,7 @@ def test_autonomous_tool_calling_loop_writes_files_and_passes(self):
resp1.content = [block_tool2]
block_final = MagicMock()
- block_final.text = "[STATUS: PASS] Solution implemented and tests pass."
+ block_final.text = "Solution implemented and tests pass.\n[STATUS: PASS]"
resp2 = MagicMock()
resp2.content = [block_final]
@@ -414,7 +414,7 @@ def fake_content(role=None, parts=None):
# Turn 1: DeepSeek finishes with PASS
part_final = MagicMock()
part_final.thought = None
- part_final.text = "[STATUS: PASS] DeepSeek solution implemented."
+ part_final.text = "DeepSeek solution implemented.\n[STATUS: PASS]"
part_final.function_call = None
resp1 = MagicMock()
@@ -422,7 +422,7 @@ def fake_content(role=None, parts=None):
cand1 = MagicMock()
cand1.content = MagicMock(parts=[part_final])
resp1.candidates = [cand1]
- resp1.text = "[STATUS: PASS] DeepSeek solution implemented."
+ resp1.text = "DeepSeek solution implemented.\n[STATUS: PASS]"
mock_client.models.generate_content.side_effect = [resp0, resp1]
mock_genai.Client.return_value = mock_client
@@ -485,7 +485,7 @@ def test_autonomous_tool_calling_loop_edit_and_read_file(self):
resp1.content = [block_edit]
block_final = MagicMock()
- block_final.text = "[STATUS: PASS] Readme edited."
+ block_final.text = "Readme edited.\n[STATUS: PASS]"
resp2 = MagicMock()
resp2.content = [block_final]
@@ -532,7 +532,7 @@ def test_mixed_manifest_picks_correct_client(self):
mock_ag.policy.allow_all = MagicMock()
mock_agent_instance = MagicMock()
mock_ag_resp = MagicMock()
- mock_ag_resp.text = AsyncMock(return_value="[STATUS: PASS] Gemini agent done")
+ mock_ag_resp.text = AsyncMock(return_value="Gemini agent done\n[STATUS: PASS]")
mock_agent_instance.chat = AsyncMock(return_value=mock_ag_resp)
mock_agent_cm = MagicMock()
mock_agent_cm.__aenter__ = AsyncMock(return_value=mock_agent_instance)
@@ -543,7 +543,7 @@ def test_mixed_manifest_picks_correct_client(self):
mock_claude_client = MagicMock()
mock_claude_resp = MagicMock()
mock_claude_block = MagicMock()
- mock_claude_block.text = "[STATUS: PASS] Claude agent done"
+ mock_claude_block.text = "Claude agent done\n[STATUS: PASS]"
mock_claude_resp.content = [mock_claude_block]
mock_claude_client.messages.create.return_value = mock_claude_resp
mock_anthropic.AnthropicVertex.return_value = mock_claude_client
@@ -551,7 +551,7 @@ def test_mixed_manifest_picks_correct_client(self):
mock_aiplatform = MagicMock()
mock_fable_endpoint = MagicMock()
mock_fable_resp = MagicMock()
- mock_fable_resp.predictions = ["[STATUS: PASS] Fable agent done"]
+ mock_fable_resp.predictions = ["Fable agent done\n[STATUS: PASS]"]
mock_fable_endpoint.predict.return_value = mock_fable_resp
mock_aiplatform.Endpoint.return_value = mock_fable_endpoint
@@ -641,7 +641,7 @@ def test_worker_task_manifest_deepseek_enables_vertex_and_global_location(self):
mock_genai = MagicMock()
mock_client = MagicMock()
mock_resp = MagicMock()
- mock_resp.text = "[STATUS: PASS] DeepSeek task done"
+ mock_resp.text = "DeepSeek task done\n[STATUS: PASS]"
mock_resp.function_calls = []
mock_resp.candidates = []
mock_client.models.generate_content.return_value = mock_resp
diff --git a/tests/test_storage_uploader.py b/tests/test_storage_uploader.py
index 09ca080..382c445 100644
--- a/tests/test_storage_uploader.py
+++ b/tests/test_storage_uploader.py
@@ -8,6 +8,7 @@
from unittest.mock import MagicMock, patch
import sys
+
sys.path.insert(0, str(Path(__file__).parent.parent / "skills" / "swarm" / "scripts"))
import storage_uploader
@@ -99,7 +100,9 @@ def test_upload_run_artifacts_generates_patch(self, mock_gen_patch, mock_upload)
self.assertIn("patch.diff", results)
self.assertTrue((self.repo_dir / "patch.diff").is_file())
- self.assertEqual((self.repo_dir / "patch.diff").read_text(encoding="utf-8"), "diff --git a/test b/test\n+new line")
+ self.assertEqual(
+ (self.repo_dir / "patch.diff").read_text(encoding="utf-8"), "diff --git a/test b/test\n+new line"
+ )
class TestGCSUploadFallback(unittest.TestCase):
@@ -198,5 +201,243 @@ def test_parse_score_metrics_missing(self):
self.assertIsNone(latency_ms)
+class TestEvidenceCredentialSafeguards(unittest.TestCase):
+ def setUp(self):
+ self.temp_dir = tempfile.TemporaryDirectory()
+ self.repo_dir = Path(self.temp_dir.name)
+ subprocess.run(["git", "init"], cwd=self.repo_dir, capture_output=True, check=True)
+ subprocess.run(["git", "config", "user.name", "Tester"], cwd=self.repo_dir, capture_output=True, check=True)
+ subprocess.run(
+ ["git", "config", "user.email", "test@test.local"], cwd=self.repo_dir, capture_output=True, check=True
+ )
+ (self.repo_dir / "app.py").write_text("print('hello')\n", encoding="utf-8")
+ subprocess.run(["git", "add", "app.py"], cwd=self.repo_dir, capture_output=True, check=True)
+ subprocess.run(["git", "commit", "-m", "init"], cwd=self.repo_dir, capture_output=True, check=True)
+
+ def tearDown(self):
+ self.temp_dir.cleanup()
+
+ @patch("storage_uploader.upload_to_gcs")
+ def test_evidence_with_credentials_rejected_and_degrades_pass(self, mock_upload):
+ uploaded_payloads = {}
+
+ def spy_upload(bucket, obj, path):
+ uploaded_payloads[obj] = Path(path).read_bytes()
+ return True
+
+ mock_upload.side_effect = spy_upload
+
+ safe_file = self.repo_dir / "safe.txt"
+ safe_bytes = b"safe measurement: 42\n"
+ safe_file.write_bytes(safe_bytes)
+
+ evidence_file = self.repo_dir / "proof.txt"
+ secret_token = "ghp_" + "a" * 36
+ cred_bytes = f"run log with sensitive token: {secret_token}\n".encode("utf-8")
+ evidence_file.write_bytes(cred_bytes)
+
+ results = storage_uploader.upload_run_artifacts(
+ bucket_name="test-bucket",
+ prefix="runs/exp1",
+ repo_dir=self.repo_dir,
+ task_index=0,
+ status="PASS",
+ evidence_files=["safe.txt", "proof.txt"],
+ )
+
+ self.assertEqual(safe_file.read_bytes(), safe_bytes)
+ self.assertEqual(evidence_file.read_bytes(), cred_bytes)
+ self.assertNotIn("evidence/proof.txt", results)
+ self.assertIn("evidence/safe.txt", results)
+ self.assertEqual(
+ uploaded_payloads.get("runs/exp1/task-0/evidence/safe.txt"),
+ safe_bytes,
+ )
+
+ secret_bytes = secret_token.encode("utf-8")
+ for obj, payload in uploaded_payloads.items():
+ self.assertNotIn(
+ secret_bytes,
+ payload,
+ f"Synthetic secret leaked in outbound upload for {obj}",
+ )
+
+ status_file = self.repo_dir / "status.json"
+ self.assertTrue(status_file.is_file())
+ status_data = json.loads(status_file.read_text(encoding="utf-8"))
+ self.assertEqual(status_data["status"], "ISSUES")
+ self.assertIn("credentials", status_data.get("evidence", ""))
+
+ @patch("storage_uploader.upload_to_gcs")
+ def test_sensitive_filenames_rejected(self, mock_upload):
+ mock_upload.return_value = True
+
+ sensitive_names = [
+ ".env",
+ "server.pem",
+ "private.key",
+ "id_rsa",
+ "id_ed25519",
+ "credentials.json",
+ "token.json",
+ ]
+ for name in sensitive_names:
+ file_path = self.repo_dir / name
+ file_path.write_text("dummy sensitive data\n", encoding="utf-8")
+ self.assertTrue(
+ storage_uploader.is_sensitive_path(file_path, self.repo_dir),
+ f"{name} should be recognized as sensitive path",
+ )
+
+ # Also verify symlink is recognized as sensitive path
+ target_file = self.repo_dir / "target.txt"
+ target_file.write_text("target", encoding="utf-8")
+ symlink_file = self.repo_dir / "symlink_file.txt"
+ symlink_file.symlink_to(target_file)
+ self.assertTrue(
+ storage_uploader.is_sensitive_path(symlink_file, self.repo_dir),
+ "Symlink should be recognized as sensitive path",
+ )
+
+ # When passed as evidence_files, all sensitive files must be rejected and status degraded
+ results = storage_uploader.upload_run_artifacts(
+ bucket_name="test-bucket",
+ prefix="runs/exp1",
+ repo_dir=self.repo_dir,
+ task_index=0,
+ status="PASS",
+ evidence_files=sensitive_names + ["symlink_file.txt"],
+ )
+
+ for name in sensitive_names:
+ self.assertNotIn(f"evidence/{name}", results)
+ self.assertNotIn("evidence/symlink_file.txt", results)
+
+ status_file = self.repo_dir / "status.json"
+ status_data = json.loads(status_file.read_text(encoding="utf-8"))
+ self.assertEqual(status_data["status"], "ISSUES")
+
+ @patch("storage_uploader.upload_to_gcs", return_value=True)
+ def test_sensitive_directory_symlink_rejected_and_degrades_pass(self, mock_upload):
+ external = self.repo_dir.parent / "external_sensitive"
+ external.mkdir(exist_ok=True)
+ (external / "leak.log").write_text("secret")
+
+ for location in ["evidence", "diagnostics"]:
+ with self.subTest(location=location):
+ status_file = self.repo_dir / "status.json"
+ if status_file.exists():
+ status_file.unlink()
+
+ evidence_files = []
+ if location == "evidence":
+ ev_dir = self.repo_dir / "ev_tree"
+ ev_dir.mkdir(exist_ok=True)
+ (ev_dir / "safe.txt").write_text("admitted bytes")
+ (ev_dir / "sym_dir").symlink_to(external)
+ evidence_files = ["ev_tree"]
+ else:
+ diag_dir = self.repo_dir / ".agystack/diagnostics"
+ diag_dir.parent.mkdir(parents=True, exist_ok=True)
+ diag_dir.symlink_to(external)
+
+ artifacts = storage_uploader.upload_run_artifacts(
+ "bucket",
+ "run",
+ self.repo_dir,
+ 0,
+ status="PASS",
+ evidence_files=evidence_files,
+ )
+ self.assertNotIn("evidence/ev_tree/sym_dir/leak.log", artifacts)
+ self.assertNotIn("evidence/.agystack/diagnostics/leak.log", artifacts)
+ status_data = json.loads(status_file.read_text(encoding="utf-8"))
+ self.assertEqual(status_data["status"], "ISSUES")
+ self.assertTrue(status_data.get("delivery_errors"))
+ self.assertEqual((external / "leak.log").read_text(), "secret")
+ if location == "evidence":
+ self.assertEqual((self.repo_dir / "ev_tree/safe.txt").read_text(), "admitted bytes")
+ self.assertIn("evidence/ev_tree/safe.txt", artifacts)
+ (self.repo_dir / "ev_tree/sym_dir").unlink()
+ (self.repo_dir / "ev_tree/safe.txt").unlink()
+ (self.repo_dir / "ev_tree").rmdir()
+ else:
+ (self.repo_dir / ".agystack/diagnostics").unlink()
+
+ @patch("storage_uploader.upload_to_gcs", return_value=True)
+ @patch("storage_uploader._generate_git_patch", return_value="replacement patch")
+ def test_patch_symlink_rejected_before_regeneration(self, mock_generate, mock_upload):
+ target = self.repo_dir / "untouched.txt"
+ target.write_bytes(b"original bytes")
+ for status, name in [("PASS", "patch.diff"), ("ISSUES", "partial.patch.diff")]:
+ with self.subTest(status=status):
+ sym_path = self.repo_dir / name
+ sym_path.symlink_to(target)
+ artifacts = storage_uploader.upload_run_artifacts(
+ "bucket",
+ "run",
+ self.repo_dir,
+ 0,
+ status,
+ ["app.py"],
+ )
+ self.assertEqual(target.read_bytes(), b"original bytes")
+ self.assertNotIn(name, artifacts)
+ self.assertEqual(json.loads((self.repo_dir / "status.json").read_text())["status"], "ISSUES")
+ sym_path.unlink()
+ mock_generate.assert_not_called()
+
+ @patch("storage_uploader.upload_to_gcs")
+ def test_failed_score_or_results_upload_degrades_pass_to_issues(self, mock_upload):
+ score_file = self.repo_dir / "score.json"
+ score_file.write_text(json.dumps({"score": 0.95}), encoding="utf-8")
+
+ tsv_file = self.repo_dir / "results.tsv"
+ tsv_file.write_text("metric\tvalue\nscore\t0.95\n", encoding="utf-8")
+
+ # 1. Test score.json upload failure
+ def fail_score_upload(bkt, dest, path):
+ if "score.json" in str(dest):
+ return False
+ return True
+
+ mock_upload.side_effect = fail_score_upload
+
+ results = storage_uploader.upload_run_artifacts(
+ bucket_name="test-bucket",
+ prefix="runs/exp1",
+ repo_dir=self.repo_dir,
+ task_index=0,
+ status="PASS",
+ )
+
+ self.assertNotIn("score.json", results)
+ status_file = self.repo_dir / "status.json"
+ status_data = json.loads(status_file.read_text(encoding="utf-8"))
+ self.assertEqual(status_data["status"], "ISSUES")
+ self.assertIn("score.json upload failed", status_data.get("evidence", ""))
+
+ # 2. Test results.tsv upload failure
+ def fail_results_upload(bkt, dest, path):
+ if "results.tsv" in str(dest):
+ return False
+ return True
+
+ mock_upload.side_effect = fail_results_upload
+
+ results = storage_uploader.upload_run_artifacts(
+ bucket_name="test-bucket",
+ prefix="runs/exp1",
+ repo_dir=self.repo_dir,
+ task_index=0,
+ status="PASS",
+ )
+
+ self.assertNotIn("results.tsv", results)
+ status_data = json.loads(status_file.read_text(encoding="utf-8"))
+ self.assertEqual(status_data["status"], "ISSUES")
+ self.assertIn("results.tsv upload failed", status_data.get("evidence", ""))
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_worker_execution_lifecycle.py b/tests/test_worker_execution_lifecycle.py
new file mode 100644
index 0000000..e2770de
--- /dev/null
+++ b/tests/test_worker_execution_lifecycle.py
@@ -0,0 +1,569 @@
+import json
+import os
+import shlex
+import subprocess
+import sys
+import time
+import types
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "skills/swarm/scripts"
+sys.path.insert(0, str(SCRIPTS))
+import cloud_dispatch
+import cloud_worker
+import result_harvester
+import storage_uploader
+from worker import partner_engine
+from worker.execution import ExecutionLimits, resolve_execution_limits
+
+
+def mock_storage(blobs=()):
+ storage = MagicMock()
+ storage.Client.return_value.bucket.return_value.list_blobs.side_effect = lambda prefix="": [
+ Blob(k, v) for k, v in (blobs.items() if isinstance(blobs, dict) else blobs) if k.startswith(prefix)
+ ]
+ g, gc = types.ModuleType("google"), types.ModuleType("google.cloud")
+ g.cloud, gc.storage = gc, storage
+ return storage, {"google": g, "google.cloud": gc, "google.cloud.storage": storage}
+
+
+class Blob:
+ def __init__(self, name, payload):
+ self.name = name
+ self.payload = payload
+
+ def download_to_filename(self, filename):
+ path = Path(filename)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(self.payload, encoding="utf-8")
+
+
+@pytest.mark.parametrize(
+ "text, expected",
+ [
+ ("Reporting `[STATUS: PASS]` now would be false. The correct status is [STATUS: BLOCKED]", "ISSUES"),
+ ("Deprecated `[STATUS: PASS]`\n[STATUS: BLOCKED]", "BLOCKED"),
+ ("Finished.\n[STATUS: PASS]", "PASS"),
+ ("[STATUS: PASS]\n[STATUS: BLOCKED]", "ISSUES"),
+ ("[STATUS: PASS]\n[STATUS: PASS]", "ISSUES"),
+ ("```text\n[STATUS: PASS]\n```", "ISSUES"),
+ ("~~~\n[STATUS: PASS]\n~~~\n[STATUS: BLOCKED]", "BLOCKED"),
+ ("> [STATUS: PASS]", "ISSUES"),
+ ('"[STATUS: PASS]"', "ISSUES"),
+ ("`[STATUS: PASS]`", "ISSUES"),
+ ("[STATUS: PASS] trailing prose", "ISSUES"),
+ ("[STATUS: DEV_IMPROVED]", "ISSUES"),
+ ("[STATUS: PASS]\nMore prose", "ISSUES"),
+ ("[STATUS: MAYBE]", "ISSUES"),
+ ("[status: issues]\n[STATUS: PASS]", "ISSUES"),
+ ("[STATUS PASS]\n[STATUS: PASS]", "ISSUES"),
+ (" [STATUS: PASS]", "ISSUES"),
+ ],
+)
+def test_strict_terminal_status(text, expected):
+ for evaluate in (cloud_worker.evaluate_agent_status, partner_engine._evaluate_agent_status):
+ assert evaluate(text)[0] == expected
+
+
+@pytest.mark.parametrize("bad", [0, -1, True, 1.5, "1.5", "oops", None])
+def test_invalid_execution_limits(bad):
+
+ with pytest.raises(ValueError):
+ resolve_execution_limits({"max_agent_turns": bad}, environ={})
+
+
+def test_limits_precedence_and_immutability():
+ from dataclasses import FrozenInstanceError
+
+ limits = resolve_execution_limits(
+ {"verify_timeout_s": 41},
+ environ={"AGYSTACK_MAX_AGENT_TURNS": "32", "COMMAND_TIMEOUT": "600"},
+ )
+ assert (limits.max_agent_turns, limits.command_timeout_s, limits.verify_timeout_s) == (32, 600, 41)
+ with pytest.raises(FrozenInstanceError):
+ limits.max_agent_turns = 2
+ with pytest.raises(ValueError):
+ resolve_execution_limits({"unexpected": 5}, environ={})
+ with pytest.raises(ValueError):
+ resolve_execution_limits([], environ={})
+ assert resolve_execution_limits(environ={}).command_timeout_s == 300
+ assert resolve_execution_limits(environ={}, command_default=900).command_timeout_s == 900
+
+
+@pytest.mark.parametrize("provider", ["claude", "fable", "deepseek"])
+@pytest.mark.parametrize("finish", [True, False])
+def test_provider_turn_limits_and_final_response(tmp_path, provider, finish):
+ (tmp_path / "note.txt").write_text("still working")
+ tool = {"type": "tool_use", "id": "read", "name": "read_file", "input": {"path": "note.txt"}}
+ responses = []
+ for turn in range(33):
+ final = finish and turn == 32
+ text = "Finished.\n[STATUS: PASS]" if final else "[STATUS: PASS]"
+ if provider == "deepseek":
+ fc = SimpleNamespace(name="read_file", args={"path": "note.txt"})
+ responses.append(SimpleNamespace(text=text, candidates=[], function_calls=[] if final else [fc]))
+ else:
+ responses.append(SimpleNamespace(content=[{"type": "text", "text": text}] + ([] if final else [tool])))
+ sdk = MagicMock()
+ if provider == "claude":
+ request = sdk.AnthropicVertex.return_value.messages.create
+ entry = partner_engine.execute_claude_model_agent
+ getter = "_get_anthropic"
+ elif provider == "fable":
+ request = sdk.Endpoint.return_value.predict
+ entry = partner_engine.execute_fable_model_agent
+ getter = "_get_aiplatform"
+ else:
+ request = sdk.Client.return_value.models.generate_content
+ entry = partner_engine.execute_deepseek_model_agent
+ getter = "_get_genai"
+ request.side_effect = responses
+ with patch.object(partner_engine, getter, return_value=sdk):
+ status, summary = entry(provider, "work", tmp_path, task_item={"execution_limits": {"max_agent_turns": 33}})
+ assert request.call_count == 33
+ assert status == ("PASS" if finish else "ISSUES")
+ if not finish:
+ assert "limit=33" in summary
+ assert "consumed=33" in summary
+ assert "read_file" in summary
+ assert "last_response" in summary
+
+
+@pytest.mark.parametrize("suffix", ["& wait", "| cat", "&"])
+def test_timeout_keeps_output_and_kills_descendants(tmp_path, suffix):
+ child = "import os,time; open('child.pid','w').write(str(os.getpid())); time.sleep(2); open('survived', 'w').write('bad')"
+ command = (
+ f"echo partial-output; echo partial-error >&2; {shlex.quote(sys.executable)} -c {shlex.quote(child)} {suffix}"
+ )
+ output = partner_engine.execute_workspace_tool(
+ "run_command",
+ {"command": command},
+ tmp_path,
+ limits=ExecutionLimits(command_timeout_s=1),
+ )
+ assert "timed out" in output.lower()
+ assert "partial-output" in output
+ assert "partial-error" in output
+ logs = list((tmp_path / ".agystack/diagnostics").glob("*.log"))
+ assert logs and "partial-output" in logs[0].read_text()
+ child_pid = int((tmp_path / "child.pid").read_text())
+ time.sleep(1.5)
+ assert not (tmp_path / "survived").exists()
+ state = subprocess.run(["ps", "-o", "stat=", "-p", str(child_pid)], capture_output=True, text=True)
+ assert not state.stdout.strip() or state.stdout.strip().startswith("Z")
+
+
+def test_command_pipeline_failure_is_not_hidden(tmp_path):
+ status, _ = cloud_worker.execute_task({"type": "command", "command": "false | tail -n 1"}, "", tmp_path, "", "")
+ assert status == "ISSUES"
+
+
+@pytest.fixture
+def worker_repo(tmp_path):
+ repo = tmp_path / "repo-task-0"
+ repo.mkdir()
+ for args in (["init"], ["config", "user.email", "test@example.com"], ["config", "user.name", "Test"]):
+ subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True)
+ (repo / "service.py").write_text("value = 1\n")
+ subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
+ subprocess.run(["git", "commit", "-m", "base"], cwd=repo, check=True, capture_output=True)
+ return repo
+
+
+@pytest.mark.parametrize(
+ "status, candidate, verify, fail_status",
+ [
+ ("BLOCKED", True, "exit 1", False),
+ ("ISSUES", True, "exit 1", False),
+ ("PASS", False, "exit 1", False),
+ ("PASS", False, "exit 0", False),
+ ("PASS", True, "exit 0", False),
+ ("PASS", True, "exit 1", False),
+ ("PASS", False, "sleep 2", False),
+ ("PASS", True, "exit 0", True),
+ ],
+)
+def test_worker_finalization_preserves_evidence(
+ worker_repo, tmp_path, monkeypatch, status, candidate, verify, fail_status, capsys
+):
+ storage = tmp_path / "storage"
+ task = {
+ "brief": "work",
+ "candidate_files": ["service.py"],
+ "evidence_files": ["results/proof.txt", "nested/domain/oracle/checks.json"],
+ "verify_command": f"echo verifier-ran > verification.txt; {verify}",
+ "execution_limits": {"verify_timeout_s": 1},
+ }
+ monkeypatch.setenv("REPO_URL", "https://example.com/repo")
+ monkeypatch.setenv("TASK_MANIFEST", json.dumps([task]))
+ monkeypatch.setenv("USE_VERTEX_AI", "1")
+ monkeypatch.setenv("GCS_BUCKET", "test-bucket")
+ monkeypatch.setenv("GCS_PREFIX", "run")
+ monkeypatch.setenv("STORAGE_MESSENGER_LOCAL_DIR", str(storage))
+ monkeypatch.chdir(tmp_path)
+ monkeypatch.setattr(cloud_worker.tempfile, "gettempdir", lambda: str(tmp_path))
+
+ raw_payload = json.dumps({"status": "verified", "checks": [1, 2, 3], "score": 0.98}, indent=2).encode("utf-8")
+
+ def clone(**kwargs):
+ repo = kwargs["repo_dir"]
+ subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
+ for key, value in [("user.email", "test@example.com"), ("user.name", "Test")]:
+ subprocess.run(["git", "config", key, value], cwd=repo, check=True)
+ (repo / "service.py").write_text("value = 1\n")
+ subprocess.run(["git", "add", "."], cwd=repo, check=True)
+ subprocess.run(["git", "commit", "-m", "base"], cwd=repo, check=True, capture_output=True)
+
+ def execute(**kwargs):
+ repo = kwargs["repo_dir"]
+ if candidate:
+ (repo / "service.py").write_text("value = 2\n")
+ (repo / "results").mkdir()
+ (repo / "results/proof.txt").write_text("partial proof")
+ deep = repo / "nested/domain/oracle/checks.json"
+ deep.parent.mkdir(parents=True, exist_ok=True)
+ deep.write_bytes(raw_payload)
+ return status, "work summary"
+
+ monkeypatch.setattr(cloud_worker, "clone_and_checkout_task", clone)
+ monkeypatch.setattr(cloud_worker, "execute_task", execute)
+ monkeypatch.setattr(cloud_worker, "StorageMessenger", None)
+
+ real_upload = storage_uploader.upload_to_gcs
+
+ def upload_hook(bucket_name, object_name, local_file_path):
+ if fail_status and object_name.endswith("status.json"):
+ return False
+ return real_upload(bucket_name, object_name, local_file_path)
+
+ monkeypatch.setattr(storage_uploader, "upload_to_gcs", upload_hook)
+
+ exit_code = 0
+ try:
+ cloud_worker.main()
+ except SystemExit as exc:
+ exit_code = exc.code
+ expected = (
+ "ISSUES"
+ if fail_status
+ else ("PASS" if status == "PASS" and verify == "exit 0" else "ISSUES" if status == "PASS" else status)
+ )
+ assert exit_code == (0 if expected == "PASS" else 1)
+ report = cloud_dispatch.parse_worker_logs(capsys.readouterr().out)
+ assert len(report) == 1 and report[0]["status"] == expected and report[0]["task_index"] == 0
+ target = storage / "test-bucket/run/task-0"
+ assert (target / "evidence/results/proof.txt").read_text() == "partial proof"
+ assert (target / "evidence/nested/domain/oracle/checks.json").read_bytes() == raw_payload
+ if fail_status:
+ assert not (target / "status.json").exists()
+ return
+
+ published = json.loads((target / "status.json").read_text())
+ assert published["status"] == expected
+ if expected == "PASS" and candidate:
+ assert (target / "patch.diff").is_file()
+ else:
+ assert not (target / "patch.diff").exists()
+ if candidate:
+ assert "+value = 2" in (target / "partial.patch.diff").read_text()
+ if status == "PASS":
+ assert (worker_repo / "verification.txt").is_file()
+ assert list((target / "evidence/.agystack/diagnostics").glob("verification-*.log"))
+ harvested = result_harvester.harvest_candidate_patches("test-bucket", "run", tmp_path / "harvest")
+ assert len(harvested) == 1
+ assert harvested[0]["status"] == published["status"]
+ assert (tmp_path / "harvest/task-0/evidence/results/proof.txt").is_file()
+ assert (tmp_path / "harvest/task-0/evidence/nested/domain/oracle/checks.json").read_bytes() == raw_payload
+ if expected == "PASS" and candidate:
+ assert harvested[0]["patch_file"] and Path(harvested[0]["patch_file"]).is_file()
+ else:
+ assert not harvested[0]["patch_file"]
+
+
+@pytest.mark.parametrize(
+ "reports",
+ [
+ [],
+ [{"task_index": 0, "status": "PASS"}],
+ [{"task_index": 0, "status": "PASS"}, {"task_index": 1, "status": "UNKNOWN"}],
+ [{"task_index": 0, "status": "PASS"}, {"task_index": 0, "status": "PASS"}],
+ ],
+)
+def test_dispatch_requires_complete_terminal_coverage(reports):
+ assert cloud_dispatch.failed_task_count(reports, 2) > 0
+
+
+def test_dispatch_complete_pass():
+ assert (
+ cloud_dispatch.failed_task_count([{"task_index": 0, "status": "PASS"}, {"task_index": 1, "status": "PASS"}], 2)
+ == 0
+ )
+
+
+@pytest.mark.parametrize(
+ "limits, bad_env",
+ [
+ (None, None),
+ ([], None),
+ ({"extra": 3}, None),
+ ({"max_agent_turns": "40"}, None),
+ ({"max_agent_turns": True}, None),
+ ({}, {"AGYSTACK_MAX_AGENT_TURNS": "oops"}),
+ ],
+)
+def test_dispatch_rejects_invalid_task_limits_before_launch(tmp_path, limits, bad_env):
+ command = [
+ sys.executable,
+ str(SCRIPTS / "cloud_dispatch.py"),
+ "--dry-run",
+ "--repo",
+ "https://example.com/repo",
+ "--vertex",
+ "--project",
+ "test",
+ "--manifest",
+ json.dumps([{"brief": "work", "execution_limits": limits}]),
+ ]
+ env = dict(os.environ, GH_TOKEN="fake-test-token")
+ if bad_env:
+ env.update(bad_env)
+ result = subprocess.run(command, cwd=tmp_path, env=env, capture_output=True, text=True)
+ assert result.returncode != 0
+ assert "execution_limits" in result.stderr or "positive integer" in result.stderr
+
+
+def test_dispatch_env_forwarding(tmp_path):
+ env = dict(os.environ, AGYSTACK_MAX_AGENT_TURNS="41", COMMAND_TIMEOUT="600", GH_TOKEN="fake-test-token")
+ env.pop("VERIFY_TIMEOUT", None)
+ result = subprocess.run(
+ [
+ sys.executable,
+ str(SCRIPTS / "cloud_dispatch.py"),
+ "--dry-run",
+ "--repo",
+ "https://example.com/repo",
+ "--vertex",
+ "--project",
+ "test",
+ "--manifest",
+ json.dumps([{"brief": "work", "execution_limits": {"max_agent_turns": 43}}]),
+ ],
+ cwd=tmp_path,
+ env=env,
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ payload = json.loads(result.stdout)
+ assert payload["env_vars"]["AGYSTACK_MAX_AGENT_TURNS"] == "41"
+ assert payload["env_vars"]["COMMAND_TIMEOUT"] == "600"
+ assert "VERIFY_TIMEOUT" not in payload["env_vars"]
+
+
+def test_deepseek_retries_consume_request_limit(tmp_path):
+ sdk = MagicMock()
+ request = sdk.Client.return_value.models.generate_content
+ request.side_effect = RuntimeError("503 unavailable")
+ with patch.object(partner_engine, "_get_genai", return_value=sdk), patch.object(partner_engine.time, "sleep"):
+ status, summary = partner_engine.execute_deepseek_model_agent(
+ "deepseek",
+ "work",
+ tmp_path,
+ task_item={"execution_limits": {"max_agent_turns": 2}},
+ )
+ assert request.call_count == 2
+ assert status == "ISSUES"
+ assert "limit=2" in summary and "consumed=2" in summary
+
+
+@pytest.mark.parametrize("provider", ["claude", "fable"])
+def test_empty_final_response_cannot_reuse_previous_pass(tmp_path, provider):
+ sdk = MagicMock()
+ tool = {"type": "tool_use", "id": "read", "name": "read_file", "input": {"path": "absent.txt"}}
+ responses = [
+ SimpleNamespace(content=[{"type": "text", "text": "[STATUS: PASS]"}, tool]),
+ SimpleNamespace(content=[]),
+ ]
+ if provider == "claude":
+ sdk.AnthropicVertex.return_value.messages.create.side_effect = responses
+ entry, getter = partner_engine.execute_claude_model_agent, "_get_anthropic"
+ else:
+ sdk.Endpoint.return_value.predict.side_effect = responses
+ entry, getter = partner_engine.execute_fable_model_agent, "_get_aiplatform"
+ with patch.object(partner_engine, getter, return_value=sdk):
+ assert entry(provider, "work", tmp_path)[0] == "ISSUES"
+
+
+def test_evidence_failure_downgrades_before_status_publication(worker_repo, tmp_path, monkeypatch):
+ (worker_repo / "service.py").write_text("value = 2\n")
+ (worker_repo / "proof.txt").write_text("proof")
+ storage = tmp_path / "storage"
+ monkeypatch.setenv("STORAGE_MESSENGER_LOCAL_DIR", str(storage))
+ published = []
+
+ def upload(bucket, name, path):
+ published.append((name, path.read_text()))
+ if name.endswith("proof.txt"):
+ return False
+ dest = storage / bucket / name
+ dest.parent.mkdir(parents=True, exist_ok=True)
+ dest.write_text(path.read_text(), encoding="utf-8")
+ return True
+
+ with patch.object(storage_uploader, "upload_to_gcs", side_effect=upload):
+ artifacts = storage_uploader.upload_run_artifacts(
+ "bucket",
+ "run",
+ worker_repo,
+ 0,
+ "PASS",
+ ["service.py"],
+ evidence_files=["proof.txt"],
+ )
+ assert published[-1][0].endswith("/status.json")
+ assert json.loads(published[-1][1])["status"] == "ISSUES"
+ assert "patch.diff" not in artifacts
+ assert "partial.patch.diff" in artifacts
+ harvested = result_harvester.harvest_candidate_patches("bucket", "run", tmp_path / "harvest")
+ assert len(harvested) == 1
+ assert harvested[0]["status"] == "ISSUES"
+ assert harvested[0]["patch_file"] is None
+ assert Path(harvested[0]["partial_patch_file"]).is_file()
+
+
+def test_missing_status_file_never_promotes_score_or_patch(tmp_path, monkeypatch):
+ source = tmp_path / "storage/bucket/run/task-0"
+ source.mkdir(parents=True)
+ (source / "score.json").write_text(json.dumps({"score": 999, "status": "PASS"}))
+ (source / "patch.diff").write_text("candidate")
+ monkeypatch.setenv("STORAGE_MESSENGER_LOCAL_DIR", str(tmp_path / "storage"))
+ results = result_harvester.harvest_candidate_patches("bucket", "run", tmp_path / "harvest")
+ assert results[0]["status"] == "UNKNOWN"
+ assert results[0]["patch_file"] is None
+
+
+def test_shell_output_is_bounded_and_credentials_scrubbed(tmp_path):
+ from worker.execution import run_bounded_shell
+
+ secret = "AIza" + "a" * 35
+ script = f"print('x' * 400000); print('{secret}'); print('last-line')"
+ result, output = run_bounded_shell(f"{shlex.quote(sys.executable)} -c {shlex.quote(script)}", tmp_path, 5)
+ assert result.returncode == 0
+ assert len(output) < 17000
+ log = next((tmp_path / ".agystack/diagnostics").glob("*.log"))
+ assert log.stat().st_size < 263000
+ assert secret not in output and secret not in log.read_text()
+ assert "[REDACTED]" in output and "last-line" in output
+
+
+@pytest.mark.parametrize("mode", ["complete", "missing", "unknown"])
+def test_attach_requires_all_worker_statuses(tmp_path, monkeypatch, mode, capsys):
+ storage = tmp_path / "storage"
+ for index in range(2 if mode != "missing" else 1):
+ directory = storage / f"bucket/swarms/run/task-{index}"
+ directory.mkdir(parents=True)
+ status = "UNKNOWN" if mode == "unknown" and index == 1 else "PASS"
+ (directory / "status.json").write_text(json.dumps({"task_index": index, "status": status}))
+ monkeypatch.chdir(tmp_path)
+ monkeypatch.setenv("STORAGE_MESSENGER_LOCAL_DIR", str(storage))
+ monkeypatch.setattr(
+ sys,
+ "argv",
+ ["cloud_dispatch.py", "--wait-execution", "execution", "--session", "run", "--gcs-bucket", "bucket"],
+ )
+ monkeypatch.setattr(cloud_dispatch, "wait_for_execution", lambda **kwargs: ({"spec": {"taskCount": 2}}, True, None))
+ monkeypatch.setenv("ANTIGRAVITY_APP_DATA_DIR", str(tmp_path / "app-data"))
+ if mode == "complete":
+ cloud_dispatch.main()
+ else:
+ with pytest.raises(SystemExit) as exc:
+ cloud_dispatch.main()
+ assert exc.value.code == 1
+
+
+def test_sdk_harvest_downloads_nested_partial_evidence(tmp_path, monkeypatch):
+ payloads = {
+ "run/task-0/status.json": json.dumps({"status": "BLOCKED"}),
+ "run/task-0/partial.patch.diff": "partial diff",
+ "run/task-0/evidence/results/proof.txt": "proof",
+ "run/task-0/evidence/.agystack/diagnostics/timeout.log": "partial output",
+ }
+
+ monkeypatch.delenv("STORAGE_MESSENGER_LOCAL_DIR", raising=False)
+ _, modules = mock_storage(payloads)
+ with patch.dict(sys.modules, modules):
+ results = result_harvester.harvest_candidate_patches("bucket", "run", tmp_path / "harvest")
+ assert results[0]["status"] == "BLOCKED"
+ assert results[0]["patch_file"] is None
+ assert Path(results[0]["partial_patch_file"]).read_text() == "partial diff"
+ assert (tmp_path / "harvest/task-0/evidence/results/proof.txt").read_text() == "proof"
+ assert (tmp_path / "harvest/task-0/evidence/.agystack/diagnostics/timeout.log").read_text() == "partial output"
+
+
+@pytest.mark.parametrize("expected_count", [None, 2])
+def test_standalone_harvest_quarantines_incomplete_coverage(tmp_path, monkeypatch, expected_count):
+ source = tmp_path / "storage/bucket/run/task-0"
+ source.mkdir(parents=True)
+ (source / "status.json").write_text(json.dumps({"status": "PASS"}))
+ (source / "patch.diff").write_text("candidate")
+ monkeypatch.setenv("STORAGE_MESSENGER_LOCAL_DIR", str(tmp_path / "storage"))
+ command = [
+ sys.executable,
+ str(SCRIPTS / "cloud_dispatch.py"),
+ "--harvest-session",
+ "run",
+ "--gcs-prefix",
+ "run",
+ "--gcs-bucket",
+ "bucket",
+ ]
+ if expected_count:
+ command += ["--tasks", str(expected_count)]
+ result = subprocess.run(command, cwd=tmp_path, capture_output=True, text=True)
+ assert result.returncode == 1
+ assert "Quarantined Candidate" in result.stdout
+ assert "Winning Candidate" not in result.stdout
+
+
+def test_harvest_candidate_patches_strict_prefix_matching(tmp_path, monkeypatch):
+ local_storage = tmp_path / "local_storage"
+ monkeypatch.setenv("STORAGE_MESSENGER_LOCAL_DIR", str(local_storage))
+ for sub, idx, score, p_text in [
+ ("canary/task-0", 0, 0.80, "patch canary"),
+ ("canary-other/task-1", 1, 0.99, "patch sibling"),
+ ]:
+ td = local_storage / "bucket/swarms" / sub
+ td.mkdir(parents=True)
+ (td / "status.json").write_text(json.dumps({"task_index": idx, "status": "PASS", "score": score}))
+ (td / "patch.diff").write_text(p_text)
+
+ local_harvest = result_harvester.harvest_candidate_patches("bucket", "swarms/canary", tmp_path / "local_harvest")
+ assert len(local_harvest) == 1
+ assert (local_harvest[0]["task_index"], local_harvest[0]["score"]) == (0, 0.80)
+ assert Path(local_harvest[0]["patch_file"]).read_text() == "patch canary"
+ assert cloud_dispatch.failed_task_count(local_harvest, 2) == 1
+
+ payloads = {
+ "run/task-0/status.json": json.dumps({"status": "PASS"}),
+ "run/task-0/score.json": json.dumps({"score": 0.95}),
+ "run/task-0/patch.diff": "valid patch",
+ "run-other/task-1/status.json": json.dumps({"status": "PASS"}),
+ "run-other/task-1/score.json": json.dumps({"score": 0.99}),
+ "run-other/task-1/patch.diff": "invalid patch",
+ "run/mytask-2/status.json": json.dumps({"status": "PASS"}),
+ "run/task-3-suffix/status.json": json.dumps({"status": "PASS"}),
+ }
+
+ monkeypatch.delenv("STORAGE_MESSENGER_LOCAL_DIR", raising=False)
+ _, modules = mock_storage(payloads)
+ with patch.dict(sys.modules, modules):
+ results = result_harvester.harvest_candidate_patches("bucket", "run", tmp_path / "harvest")
+
+ assert len(results) == 1
+ assert results[0]["task_index"] == 0
+ assert results[0]["status"] == "PASS"
+ assert results[0]["score"] == 0.95