File size: 14,268 Bytes
5090f42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 | """Private JSONL adapter for the pinned upstream Argus 0.1.1 Python service.
Automaker owns this one long-lived process. Direct service calls intentionally avoid the WebAPI
message route because that route may start Argus's native Engineer.
"""
import asyncio, importlib, inspect, json, os, shutil, sys
projects, events = {}, {}
service = manager_bridge = None
fake = os.environ.get("ARGUS_BRIDGE_FAKE") == "1"
class BridgeFailure(Exception):
def __init__(self, code, message, retryable=False):
super().__init__(message); self.code, self.retryable = code, retryable
def load_service():
global service, manager_bridge
if fake or service is not None: return service
try:
cls = getattr(importlib.import_module("argus_skill.plugin.service"), "ArgusPluginService")
manager_bridge = importlib.import_module("argus_skill.webapi.manager_bridge")
service = cls()
return service
except (ImportError, AttributeError, TypeError) as exc:
raise BridgeFailure("ARGUS_UPSTREAM_CONTRACT_MISMATCH", type(exc).__name__)
def await_value(value):
return asyncio.run(value) if inspect.isawaitable(value) else value
def invoke_manager(name, **kwargs):
load_service()
target = getattr(manager_bridge, name, None)
if not callable(target):
raise BridgeFailure("ARGUS_UPSTREAM_CONTRACT_MISMATCH", f"manager_bridge.{name} is unavailable")
return await_value(target(**kwargs))
def plain(value):
if hasattr(value, "model_dump"): return value.model_dump()
if hasattr(value, "__dict__"): return dict(value.__dict__)
return value if isinstance(value, dict) else {"value": value}
def parse_chat_json(response, code):
response = plain(response)
if response.get("kind") != "chat" or not isinstance(response.get("reply"), str):
raise BridgeFailure(code, "Manager did not return a chat reply")
reply = response["reply"].strip()
if reply.startswith("```"):
lines = reply.splitlines()
if lines and lines[0].strip().lower() in ("```", "```json") and lines[-1].strip() == "```":
reply = "\n".join(lines[1:-1]).strip()
try: return json.loads(reply)
except (TypeError, json.JSONDecodeError): raise BridgeFailure(code, "Manager chat reply was not strict JSON")
def validate_decision(value):
if not isinstance(value, dict) or not isinstance(value.get("admitted"), bool) or not isinstance(value.get("reason"), str):
raise BridgeFailure("ARGUS_MANAGER_FAILED", "Manager decision schema is invalid")
if value.get("recommendedExecutor") not in ("junior", "senior", "review_only", "blocked"):
raise BridgeFailure("ARGUS_MANAGER_FAILED", "Manager executor is invalid")
if not isinstance(value.get("evidence"), list) or not all(isinstance(item, str) for item in value["evidence"]) or not isinstance(value.get("needsNextSpec"), bool):
raise BridgeFailure("ARGUS_MANAGER_FAILED", "Manager decision schema is invalid")
return value
def validate_evaluation(value):
if not isinstance(value, dict) or not isinstance(value.get("satisfied"), bool) or not isinstance(value.get("reasoning"), str):
raise BridgeFailure("OBJECTIVE_EVALUATION_FAILED", "Objective evaluation schema is invalid")
if not isinstance(value.get("evidence"), list) or not all(isinstance(item, str) for item in value["evidence"]):
raise BridgeFailure("OBJECTIVE_EVALUATION_FAILED", "Objective evidence is invalid")
gaps = value.get("remainingGaps")
if not isinstance(gaps, list) or any(not isinstance(g, dict) or not isinstance(g.get("title"), str) or not isinstance(g.get("description"), str) or not isinstance(g.get("evidence"), list) or not all(isinstance(item, str) for item in g["evidence"]) for g in gaps):
raise BridgeFailure("OBJECTIVE_EVALUATION_FAILED", "Remaining gaps schema is invalid")
return value
def task_from_row(row, revision):
task_id = str(row.get("id") or row.get("task_id") or "")
return {"taskId": task_id, "title": row.get("title") or row.get("objective") or task_id,
"objective": row.get("objective") or row.get("title") or "",
"description": row.get("description") or row.get("reason") or "",
"acceptanceCriteria": row.get("acceptance_criteria") or ([row["acceptance_check"]] if row.get("acceptance_check") else []),
"evidenceRequired": row.get("evidence_required") or row.get("context_refs") or [],
"recommendedExecutor": "junior", "dependencies": row.get("deps") or [],
"risk": row.get("risk") or "normal", "sourceSpecRevision": revision}
def operate(operation, params):
if operation == "health":
binary = os.environ.get("ARGUS_BIN") or shutil.which("argus")
return {"running": True, "upstreamInstalled": bool(binary) or fake,
"version": "fake-0.1.1" if fake else "0.1.1",
"home": os.environ.get("ARGUS_SKILL_HOME"),
"provider": os.environ.get("ARGUS_SKILL_OPENCODE_PROVIDER"),
"model": os.environ.get("ARGUS_SKILL_MODEL"),
"roleBackends": {role: os.environ.get(f"ARGUS_SKILL_{role}_BACKEND")
for role in ("MANAGER", "PLANNER", "ENGINEER", "REVIEWER")}}
if operation == "shutdown":
return {"stopped": True}
pid = params.get("projectId", "")
if operation == "ensure_project":
created = pid not in projects
persisted_sid = params.get("nativeProjectId")
if persisted_sid:
params = dict(params); params["nativeSid"] = persisted_sid
elif fake:
params = dict(params); params["nativeSid"] = f"native-{pid}"
else:
target = getattr(load_service(), "create_project", None)
if not callable(target): raise BridgeFailure("ARGUS_UPSTREAM_CONTRACT_MISMATCH", "create_project is unavailable")
result = plain(await_value(target(workdir=params["projectPath"], name=pid)))
if result.get("ok") is not True:
raise BridgeFailure("ARGUS_PROJECT_CREATE_FAILED", str(result.get("error") or result.get("message") or "create_project failed"))
project = plain(result.get("project"))
native_sid = str(project.get("sid") or "")
if not native_sid: raise BridgeFailure("ARGUS_PROJECT_CREATE_FAILED", "create_project returned no SID")
params = dict(params); params["nativeSid"] = native_sid
projects.setdefault(pid, dict(params)); events.setdefault(pid, [])
return {"projectId": projects[pid].get("nativeSid", pid), "correlationId": pid, "created": created}
if operation == "resume_project":
return {"projectId": projects.get(pid, {}).get("nativeSid", pid), "status": "resumed"}
if operation == "get_status": return {"status": "ready", "projectId": projects.get(pid, {}).get("nativeSid", pid)}
if operation == "manager_handoff":
if fake:
result = {"admitted": True, "reason": "campaign advances the objective", "evidence": ["ARGUS.md", "spec.md"], "recommendedExecutor": "junior", "needsNextSpec": False}
projects[pid]["decision"] = result; return result
sid = projects[pid]["nativeSid"]
prompt = ("Act only as the Automaker admission brain. Do not enqueue a mission or execute work. "
"Using the immutable objective and current campaign below, return ONLY JSON with admitted, reason, evidence, recommendedExecutor, and needsNextSpec.\n\n"
f"ARGUS.md:\n{params.get('objective','')[:30000]}\n\nspec.md:\n{params.get('spec','')[:30000]}\n\nLifecycle:\n{json.dumps(params.get('lifecycle',{}))}")
response = invoke_manager("manager_message", sid=sid, text=prompt, global_root=params.get("projectPath"), route_override="chat", source_channel="automaker")
return validate_decision(parse_chat_json(response, "ARGUS_MANAGER_FAILED"))
if operation == "planner_next_task":
revision = int(params.get("sourceSpecRevision", 1))
if fake:
task = {"taskId": f"argus-task-r{revision}", "title": "Implement current specification", "objective": "Advance the current campaign", "description": "Implement the next bounded specification unit", "acceptanceCriteria": ["Relevant tests pass"], "evidenceRequired": ["Test output"], "recommendedExecutor": "junior", "dependencies": [], "risk": "normal", "sourceSpecRevision": revision}
projects[pid]["tasks"] = [task]; return task
brief = ("Create ONE bounded next implementation task. Deployment, release, and publishing are prohibited. "
"Include concrete acceptance and evidence expectations.\n\n"
f"Stable objective:\n{params.get('objective','')[:30000]}\n\nCurrent spec:\n{params.get('spec','')[:30000]}\n\n"
f"Manager decision:\n{json.dumps(params.get('managerDecision',{}))}\nSpec revision: {revision}\nAdmission context: {json.dumps(params.get('context',{}))}")
plan = plain(invoke_manager("manager_plan", sid=projects[pid]["nativeSid"], text=brief, global_root=params.get("projectPath")))
if plan.get("error"): raise BridgeFailure("ARGUS_PLANNER_FAILED", str(plan["error"]), True)
steps = plan.get("steps") or []
if not steps: return None
row = steps[0] if isinstance(steps[0], dict) else {"title": str(steps[0]), "objective": str(steps[0])}
row.setdefault("id", f"planner-r{revision}-1"); row.setdefault("description", str(plan.get("notes") or ""))
return task_from_row(row, revision)
if operation == "get_pending_work": return projects.get(pid, {}).get("tasks", [])
if operation == "get_recent_events": return events.get(pid, [])[-20:]
if operation == "evaluate_objective":
if fake:
return {"satisfied": False, "reasoning": "remaining work", "evidence": ["completed task"],
"remainingGaps": [{"title": "Next gap", "description": "Continue implementation", "evidence": []}]}
prompt = ("Evaluate whether the immutable ARGUS.md objective is fully satisfied. Do not enqueue work and do not redefine the objective. "
"Return ONLY strict JSON with satisfied, reasoning, evidence, and remainingGaps.\n\n"
f"ARGUS.md:\n{params.get('objective','')[:30000]}\n\nspec.md:\n{params.get('spec','')[:30000]}\n\nCompleted evidence:\n{json.dumps(params.get('evidence',{}))}")
response = invoke_manager("manager_message", sid=projects[pid]["nativeSid"], text=prompt, global_root=projects[pid].get("projectPath"), route_override="chat", source_channel="automaker")
return validate_evaluation(parse_chat_json(response, "OBJECTIVE_EVALUATION_FAILED"))
if operation == "renew_specification":
if fake:
return {"spec": "# Renewed campaign\n\n## What was achieved\nPrevious task completed.\n\n## What remains\nNext gap.\n\n## Acceptance criteria\n- Complete the next gap.\n\n## Non-goals\n- Deployment.\n"}
brief = ("Create the next implementation campaign that closes the remaining objective gaps. Return structured steps and notes; do not execute.\n\n"
f"Immutable ARGUS.md:\n{params.get('objective','')[:30000]}\n\nPrevious spec:\n{params.get('spec','')[:30000]}\n\n"
f"Objective evaluation:\n{json.dumps(params.get('evaluation',{}))}\n\nNon-goals: deployment, release, publishing.")
plan = plain(invoke_manager("manager_plan", sid=projects[pid]["nativeSid"], text=brief, global_root=projects[pid].get("projectPath")))
if plan.get("error"): raise BridgeFailure("SPEC_RENEWAL_FAILED", str(plan["error"]))
steps = plan.get("steps")
if not isinstance(steps, list) or not steps: raise BridgeFailure("SPEC_RENEWAL_FAILED", "Planner returned no steps")
evaluation = params.get("evaluation", {})
achieved = "\n".join(f"- {item}" for item in evaluation.get("evidence", [])) or "- Previous campaign completed."
gaps = "\n".join(f"- **{g.get('title','Gap')}**: {g.get('description','')}" for g in evaluation.get("remainingGaps", [])) or "- None recorded."
planned = []
for index, step in enumerate(steps, 1):
row = step if isinstance(step, dict) else {"title": str(step), "detail": str(step)}
planned.append(f"### {row.get('title') or f'Step {index}'}\n{row.get('detail') or row.get('description') or row.get('objective') or ''}")
acceptance = [str(item) for step in steps if isinstance(step, dict) for item in (step.get("acceptance_criteria") or step.get("acceptanceCriteria") or [])]
evidence = [str(item) for step in steps if isinstance(step, dict) for item in (step.get("evidence_required") or step.get("evidenceRequired") or [])]
spec = (f"# Campaign {params.get('sourceSpecRevision', '')}\n\n## Context\n{evaluation.get('reasoning','Further objective work is required.')}\n\n"
f"## What was achieved\n{achieved}\n\n## Remaining objective gaps\n{gaps}\n\n## Planned work\n" + "\n\n".join(planned) +
"\n\n## Acceptance criteria\n" + ("\n".join(f"- {x}" for x in acceptance) or "- Complete and verify the planned work.") +
"\n\n## Evidence required\n" + ("\n".join(f"- {x}" for x in evidence) or "- Relevant test output.") +
"\n\n## Non-goals\n- Deployment\n- Release\n- Publishing\n")
return {"spec": spec}
raise BridgeFailure("ARGUS_UNKNOWN_OPERATION", operation)
for raw in sys.stdin:
request = {"requestId": 0}
try:
request = json.loads(raw); result = operate(request.get("operation", ""), request.get("params", {}))
response = {"requestId": request["requestId"], "ok": True, "result": result}
except BridgeFailure as exc:
response = {"requestId": request["requestId"], "ok": False, "error": {"code": exc.code, "message": str(exc), "retryable": exc.retryable}}
except Exception as exc:
response = {"requestId": request["requestId"], "ok": False, "error": {"code": "ARGUS_MALFORMED_RESPONSE", "message": f"{type(exc).__name__}: {str(exc)[:200]}", "retryable": False}}
print(json.dumps(response), flush=True)
if request.get("operation") == "shutdown": break
|