feat: added supervisor mode

harness/investigators.py

@@ -0,0 +1,101 @@
+import json
+from dataclasses import dataclass
+
+from openai import AsyncOpenAI
+from openai.types.responses import FunctionToolParam
+
+from config import settings
+from harness.tools import TOOL_SCHEMAS, get_charges, search_knowledge_base
+
+client = AsyncOpenAI(api_key=settings.openai_api_key)
+
+MODEL = "gpt-5.6-luna"
+MAX_STEPS = 5
+
+# Reuse the shared schema
+_SEARCH_KB = next(s for s in TOOL_SCHEMAS if s["name"] == "searchKnowledgeBase")
+
+# getCharges has NO shared schema on purpose - defining it here keeps it inside the
+# read-only investigator subsystem instead of widening the main agent's surface.
+_GET_CHARGES: FunctionToolParam = {
+    "type": "function",
+    "strict": False,
+    "name": "getCharges",
+    "description": "Look up a customer's charges.",
+    "parameters": {
+        "type": "object",
+        "properties": {"customerId": {"type": "string"}},
+        "required": ["customerId"],
+    },
+}
+
+_TOOL_FNS = {
+    # call.arguments == '{"customerId": "cus_88121"}'
+    "getCharges": lambda a: get_charges(a.get("customerId", "")),
+    "searchKnowledgeBase": lambda a: search_knowledge_base(a.get("query", "")),
+}
+
+
+@dataclass(frozen=True)
+class Investigator:
+    system_prompt: str
+    tools: tuple[FunctionToolParam, ...]
+
+
+INVESTIGATORS: dict[str, Investigator] = {
+    "billing": Investigator(
+        system_prompt="You are a billing investigator. Use getCharges to find duplicate or "
+        "erroneous charges. Report the charge ids, the amount, and the refund "
+        "you'd recommend — concisely.",
+        tools=(_GET_CHARGES,),
+    ),
+    "technical": Investigator(
+        system_prompt="You are a technical investigator. Use searchKnowledgeBase to find known "
+        "bugs and workarounds. Report the issue, any ticket, and the workaround — "
+        "concisely.",
+        tools=(_SEARCH_KB,),
+    ),
+    "sales": Investigator(
+        system_prompt="You are a sales investigator. Use searchKnowledgeBase for pricing "
+        "guidance, then state the relevant numbers and next step — concisely.",
+        tools=(_SEARCH_KB,),
+    ),
+}
+
+
+async def run_investigator(agent: str, objective: str):
+    """Run one investigator over its objective in its OWN context. Returns findings."""
+    investigator = INVESTIGATORS.get(agent)
+    if investigator is None:
+        raise ValueError(f"unknown investigator: {agent}")
+
+    # Turn 1 is a plain inference: system via `instructions`, objective via `input`.
+    input_items: str | list = objective
+    for _ in range(MAX_STEPS):
+        resp = await client.responses.create(
+            model=MODEL,
+            instructions=investigator.system_prompt,
+            input=input_items,
+            tools=investigator.tools,
+        )
+        calls = [item for item in resp.output if item.type == "function_call"]
+        if not calls:
+            return resp.output_text
+
+        # A tool fired -> upgrade to the list form so we can feed the result back.
+        if isinstance(input_items, str):
+            input_items = [{"role": "user", "content": input_items}]
+        input_items += [item.model_dump(exclude={"status"}) for item in resp.output]
+        for call in calls:
+            result = _TOOL_FNS[call.name](json.loads(call.arguments))
+            input_items.append(
+                {
+                    "type": "function_call_output",
+                    "call_id": call.call_id,
+                    "output": json.dumps(result),
+                }
+            )
+    # Exhausted MAX_STEPS with the model still calling tools. Raise so the
+    # supervisor's fan-in records subagent.failed instead of a fake-successful
+    # empty finding (which would make synthesis say "still pending").
+    raise RuntimeError(f"investigator {agent!r} hit the {MAX_STEPS}-step limit")

harness/supervisor.py

@@ -0,0 +1,127 @@
+# harness/supervisor.py
+import asyncio
+from typing import Literal
+
+from dbos import DBOS
+from pydantic import BaseModel
+
+from harness.bus import emit
+from harness.investigators import run_investigator
+from harness.runtime import client, emit_step  # reuse both; also forces DBOS() first
+
+MODEL = "gpt-5.6-luna"
+MAX_PLAN_STEPS = 6  # cap concurrent investigators — steps come from model output
+
+
+# The PLAN is a first-class artifact: a structured object the supervisor emits,
+# the inspector renders, the synthesis reads, and that survives a crash. It's
+# untrusted model output crossing a trust boundary → a Pydantic model is right.
+class PlanStep(BaseModel):
+    id: str
+    agent: Literal["billing", "technical", "sales"]
+    objective: str
+
+
+class Plan(BaseModel):
+    steps: list[PlanStep]
+
+
+@DBOS.step()
+async def plan_step(task: str) -> list[dict]:
+    # Structured output: the model must return something shaped like Plan.
+    resp = await client.responses.parse(
+        model=MODEL,
+        instructions=(
+            "Decompose a customer escalation into independent sub-tasks — one per "
+            "area the message actually raises (billing / technical / sales). Only "
+            "include relevant areas."
+        ),
+        input=task,
+        text_format=Plan,
+    )
+    plan = resp.output_parsed
+    return [step.model_dump() for step in plan.steps] if plan else []
+
+
+@DBOS.step()
+async def investigate_step(step: dict, workflow_id: str) -> dict:
+    # started/completed are emitted INSIDE the step (like tool_step does).
+    emit(
+        {
+            "type": "subagent.started",
+            "workflowId": workflow_id,
+            "stepId": step["id"],
+            "agent": step["agent"],
+            "objective": step["objective"],
+        }
+    )
+    findings = await run_investigator(step["agent"], step["objective"])
+    emit(
+        {
+            "type": "subagent.completed",
+            "workflowId": workflow_id,
+            "stepId": step["id"],
+            "agent": step["agent"],
+            "findings": findings,
+        }
+    )
+    return {"agent": step["agent"], "findings": findings}
+
+
+@DBOS.step()
+async def synthesize_step(task: str, findings: list[dict]) -> str:
+    joined = "\n\n".join(f"[{f['agent']}] {f['findings']}" for f in findings) or "(none)"
+    resp = await client.responses.create(
+        model=MODEL,
+        instructions=(
+            "You are a support lead. Using your investigators' findings, write ONE "
+            "clear, friendly reply to the customer that addresses every point they "
+            "raised. If an area's investigation is missing, acknowledge it briefly "
+            "and say you'll follow up."
+        ),
+        input=f"Customer escalation:\n{task}\n\nInvestigator findings:\n{joined}",
+    )
+    return resp.output_text
+
+
+# THE SUPERVISOR. Plan → dispatch sub-agents in parallel → fan in → synthesize.
+# Unlike a handoff, the supervisor keeps control the whole time.
+@DBOS.workflow()
+async def supervisor_workflow(task: str) -> str:
+    workflow_id = DBOS.workflow_id or "unknown"
+    await emit_step({"type": "workflow.started", "workflowId": workflow_id, "input": task})
+
+    # PLAN
+    steps = await plan_step(task)
+    await emit_step({"type": "plan.created", "workflowId": workflow_id, "steps": steps})
+
+    # DISPATCH — every sub-agent runs in parallel, each in its own context window.
+    # `steps` is untrusted model output; cap it so a degenerate plan can't spawn
+    # unbounded concurrent investigators (each up to MAX_STEPS model calls).
+    steps = steps[:MAX_PLAN_STEPS]
+    results = await asyncio.gather(
+        *(investigate_step(step, workflow_id) for step in steps),
+        return_exceptions=True,
+    )
+
+    # FAN-IN — keep successes, record failures, and keep going (degrade).
+    findings: list[dict] = []
+    for step, result in zip(steps, results):
+        if isinstance(result, BaseException):
+            await emit_step(
+                {
+                    "type": "subagent.failed",
+                    "workflowId": workflow_id,
+                    "stepId": step["id"],
+                    "agent": step["agent"],
+                    "error": str(result),
+                }
+            )
+        else:
+            findings.append(result)
+
+    # SYNTHESIZE
+    reply = await synthesize_step(task, findings)
+    await emit_step({"type": "model.completed", "workflowId": workflow_id, "text": reply})
+    await emit_step({"type": "workflow.completed", "workflowId": workflow_id, "output": reply})
+    return reply

server.py

@@ -10,6 +10,7 @@
 from harness.code_mode import call
 from harness.db import db_client, ensure_schema
 from harness.runtime import agent_workflow
+from harness.supervisor import supervisor_workflow
 
 logger = logging.getLogger("harness.rpc")
 
@@ -29,10 +30,13 @@ async def lifespan(app: FastAPI):
 _running_tasks: set[asyncio.Task] = set()
 
 
-async def run_task(task: str) -> None:
+async def run_task(task: str, mode: str = "default") -> None:
     workflow_id = ""
     try:
-        handle = await DBOS.start_workflow_async(agent_workflow, task)
+        if mode == "supervised":
+            handle = await DBOS.start_workflow_async(supervisor_workflow, task)
+        else:
+            handle = await DBOS.start_workflow_async(agent_workflow, task)
         workflow_id = handle.workflow_id
         await handle.get_result()
     except Exception as e:
@@ -83,9 +87,10 @@ async def forward():
             if msg.get("type") == "submit_task":
                 task = msg.get("input")
                 if task:
+                    mode = msg.get("mode", "default")
                     # Run the agent in the background (agent -> queue), keeping a
                     # reference so the task isn't garbage-collected mid-run.
-                    t = asyncio.create_task(run_task(task))
+                    t = asyncio.create_task(run_task(task, mode))
                     _running_tasks.add(t)
                     t.add_done_callback(_running_tasks.discard)
     except WebSocketDisconnect: