"""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