Deploy from 40e5ccd

app/harness/events.py

@@ -0,0 +1,54 @@
+"""In-process fan-out from a detached run to however many browsers are watching.
+
+Deliberately not durable: the durable record is `skill_run_steps`, and the
+stream endpoint replays from there before it follows this bus. That split is
+what makes a disconnect survivable — the SSE path reads *from* the run record
+rather than *being* the run.
+"""
+
+import asyncio
+import logging
+
+log = logging.getLogger(__name__)
+
+# Bounded so a browser that stops reading cannot grow memory without limit.
+_MAX_QUEUE = 256
+
+DONE = object()
+"""Sentinel pushed by close(): end of stream, not an event."""
+
+_subscribers: dict[int, set[asyncio.Queue]] = {}
+
+
+def subscribe(run_id: int) -> asyncio.Queue:
+    queue: asyncio.Queue = asyncio.Queue(maxsize=_MAX_QUEUE)
+    _subscribers.setdefault(run_id, set()).add(queue)
+    return queue
+
+
+def unsubscribe(run_id: int, queue: asyncio.Queue) -> None:
+    listeners = _subscribers.get(run_id)
+    if listeners is None:
+        return
+    listeners.discard(queue)
+    if not listeners:
+        _subscribers.pop(run_id, None)
+
+
+def publish(run_id: int, event: dict) -> None:
+    """Non-blocking by contract. A stalled reader loses events; it must never
+    stall the run producing them, and it never loses the durable step rows."""
+    for queue in list(_subscribers.get(run_id, ())):
+        try:
+            queue.put_nowait(event)
+        except asyncio.QueueFull:
+            log.warning("dropping event for run %s: subscriber queue full", run_id)
+
+
+def close(run_id: int) -> None:
+    for queue in list(_subscribers.get(run_id, ())):
+        try:
+            queue.put_nowait(DONE)
+        except asyncio.QueueFull:
+            pass
+    _subscribers.pop(run_id, None)

app/harness/executor.py

@@ -0,0 +1,116 @@
+"""Starting a run and letting it outlive the request that asked for it.
+
+`asyncio.create_task` on the same event loop — no worker process, no queue,
+matching the single-instance Space. Closing the browser does not kill the run.
+"""
+
+import asyncio
+import logging
+
+from pydantic import ValidationError
+
+from app.harness import events, runner, runs, tools
+from app.services import sync
+
+log = logging.getLogger(__name__)
+
+# Strong references, keyed by run id. The event loop only holds a weak one, so
+# a task not kept here can be garbage-collected mid-run; keying by run id is
+# also what lets `cancel` find the task to stop.
+_tasks: dict[int, asyncio.Task] = {}
+
+
+class InvalidInput(Exception):
+    """The submitted input did not match the skill's declared input model."""
+
+
+def build_registry() -> tools.ToolRegistry:
+    """The tools a run may call. PR 3 fills this in with the real builtins."""
+    return tools.ToolRegistry()
+
+
+async def start(skill, inputs: dict, model: str | None, owner: bool) -> dict:
+    """Validate, create the run row, detach the work, return immediately.
+
+    Input validation happens BEFORE the row exists so a malformed submission
+    leaves no failed run cluttering history — and, more importantly, does not
+    consume the one-active-run slot.
+    """
+    try:
+        validated = skill.input_model.model_validate(inputs)
+    except ValidationError as exc:
+        raise InvalidInput(str(exc)) from exc
+
+    payload = validated.model_dump()
+    run = runs.create_run(skill.name, skill.scheduler.name, model, payload)
+    task = asyncio.create_task(execute(skill, run, payload, owner))
+    _tasks[run["id"]] = task
+    task.add_done_callback(lambda _t: _tasks.pop(run["id"], None))
+    return run
+
+
+async def execute(skill, run: dict, inputs: dict, owner: bool) -> None:
+    """The detached body. Every exit path writes a terminal row — an exception
+    escaping here would vanish into the loop's handler and leave the row at
+    'running' until the next boot's sweep."""
+    run_id = run["id"]
+    runs.start_run(run_id)
+    events.publish(run_id, {"type": "run-start", "run_id": run_id, "skill": skill.name})
+    ctx = runner.RunContext(
+        run_id=run_id, model=run["model"], owner=owner,
+        inputs=inputs, registry=build_registry(),
+    )
+    try:
+        output = await skill.scheduler.run(ctx)
+        runs.finish_run(run_id, output or {})
+        events.publish(
+            run_id, {"type": "run-done", "run_id": run_id, "output": output or {}}
+        )
+    except runner.StepFailure as exc:
+        runs.fail_run(run_id, exc.code, exc.message)
+        event = {"type": "error", "code": exc.code, "message": exc.message}
+        if exc.retry_after is not None:
+            event["retry_after"] = exc.retry_after
+        events.publish(run_id, event)
+    except asyncio.CancelledError:
+        # CancelledError is a BaseException, so the broad `except Exception`
+        # below would NOT catch it — without this the run row would be left at
+        # 'running' until the next boot's sweep. Re-raised after recording, so
+        # asyncio still sees a properly cancelled task.
+        runs.cancel_run(run_id)
+        events.publish(run_id, {"type": "error", "code": "cancelled",
+                                "message": "The run was cancelled."})
+        raise
+    except Exception as exc:
+        log.exception("run %s failed unexpectedly", run_id)
+        runs.fail_run(run_id, "internal_error", str(exc))
+        events.publish(
+            run_id,
+            {"type": "error", "code": "internal_error", "message": "The run failed."},
+        )
+    finally:
+        events.close(run_id)
+        # Once, at the terminal state: the 30s debounce means pushing per-step
+        # would churn the dataset repo for no gain, but a finished run that is
+        # never pushed is lost to the next restart.
+        sync.schedule_push()
+
+
+async def drain() -> None:
+    """Await every in-flight run. Test helper; also useful for a clean shutdown."""
+    while _tasks:
+        await asyncio.gather(*list(_tasks.values()), return_exceptions=True)
+
+
+def cancel(run_id: int) -> bool:
+    """Stop an in-flight run. Returns False when there is nothing to stop.
+
+    The row is written by `execute`'s CancelledError handler rather than here,
+    so cancellation records itself through the same path whether it is
+    requested or arrives some other way.
+    """
+    task = _tasks.get(run_id)
+    if task is None or task.done():
+        return False
+    task.cancel()
+    return True

app/harness/runner.py

@@ -0,0 +1,349 @@
+"""The instrumented middle: what a step is, what a model call costs, and what
+gets written down about both.
+
+Every model call and every tool call goes through here, which is the only
+reason the run record can be trusted as the source of truth for the live view
+and for v0.8.0's analytics.
+"""
+
+import json
+import time
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from typing import Any
+
+from app.config import get_settings
+from app.harness import contracts, events, runs, tools
+from app.services import openrouter
+
+
+class StepFailure(Exception):
+    """A step could not produce valid output. Fails the run, recording which
+    step and why — no partial output is ever filed as a proposal."""
+
+    def __init__(self, code: str, message: str, retry_after: int | None = None):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.retry_after = retry_after
+
+
+class BudgetExceeded(StepFailure):
+    def __init__(self, used: int, allowed: int):
+        super().__init__(
+            "budget_exceeded",
+            f"This run reached its {allowed}-call limit (used {used}).",
+        )
+
+
+@dataclass
+class _OpenStep:
+    step_id: int
+    ordinal: int
+    started: float
+    model: str | None = None
+    tokens_in: int = 0
+    tokens_out: int = 0
+
+
+@dataclass
+class RunContext:
+    run_id: int
+    model: str | None
+    owner: bool
+    inputs: dict
+    registry: tools.ToolRegistry
+    max_calls: int | None = None
+    retries: int | None = None
+    state: dict = field(default_factory=dict)
+    calls_used: int = 0
+    _open: _OpenStep | None = None
+
+    def __post_init__(self) -> None:
+        settings = get_settings()
+        if self.max_calls is None:
+            self.max_calls = settings.skill_max_model_calls
+        if self.retries is None:
+            self.retries = settings.skill_contract_retries
+
+    def emit(self, event: dict) -> None:
+        events.publish(self.run_id, event)
+
+    @asynccontextmanager
+    async def step(self, name: str):
+        """Open a step: one row, one start event, one done/failed event.
+
+        Steps that make no model call still get a row — `gather` is a real
+        step the user waits on, and a timeline that skips it is wrong.
+        """
+        step_id = runs.add_step(self.run_id, name)
+        ordinal = len(runs.list_steps(self.run_id))
+        open_step = _OpenStep(
+            step_id=step_id, ordinal=ordinal, started=time.monotonic()
+        )
+        self._open = open_step
+        self.emit({"type": "step-start", "name": name, "ordinal": ordinal})
+        try:
+            yield open_step
+        except StepFailure as exc:
+            self._close_step(open_step, status="failed", error=exc.code)
+            self.emit({
+                "type": "step-done", "name": name, "ordinal": ordinal,
+                "status": "failed", "error": exc.code,
+            })
+            raise
+        except Exception as exc:
+            self._close_step(open_step, status="failed", error=type(exc).__name__)
+            self.emit({
+                "type": "step-done", "name": name, "ordinal": ordinal,
+                "status": "failed", "error": type(exc).__name__,
+            })
+            raise
+        else:
+            latency = self._close_step(open_step, status="succeeded")
+            self.emit({
+                "type": "step-done", "name": name, "ordinal": ordinal,
+                "status": "succeeded", "latency_ms": latency,
+                "tokens_in": open_step.tokens_in or None,
+                "tokens_out": open_step.tokens_out or None,
+            })
+
+    def _close_step(
+        self, open_step: _OpenStep, *, status: str, error: str | None = None
+    ) -> int:
+        if self._open is open_step:
+            self._open = None
+        latency = int((time.monotonic() - open_step.started) * 1000)
+        runs.finish_step(
+            open_step.step_id,
+            status=status,
+            model=open_step.model,
+            tokens_in=open_step.tokens_in or None,
+            tokens_out=open_step.tokens_out or None,
+            latency_ms=latency,
+            error=error,
+        )
+        return latency
+
+    async def call_model(
+        self,
+        messages: list[dict],
+        contract: "contracts.TextContract | contracts.JsonContract",
+        *,
+        tool_definitions: list[dict] | None = None,
+    ) -> Any:
+        """One contract-validated completion, retried a bounded number of times.
+
+        Provider-side enforcement first (`response_format`), extraction second
+        (inside the contract), and only then a retry that quotes the exact
+        validation failure back.
+        """
+        conversation = list(messages)
+        attempts = (self.retries or 0) + 1
+        last_error = ""
+        for _ in range(attempts):
+            raw = await self._one_completion(conversation, contract, tool_definitions)
+            content = raw.get("content") or ""
+            try:
+                return contract.validate(content)
+            except contracts.ContractError as exc:
+                last_error = str(exc)
+                conversation = conversation + [
+                    {"role": "assistant", "content": content},
+                    contracts.repair_message(last_error),
+                ]
+        raise StepFailure(
+            "contract_invalid",
+            f"The model's output did not match the required format: {last_error}",
+        )
+
+    async def call_model_raw(
+        self, messages: list[dict], *, tool_definitions: list[dict] | None = None
+    ) -> dict:
+        """A completion with no contract — the agent scheduler's tool loop,
+        where the decision is 'which tool', not 'what shape'."""
+        return await self._one_completion(
+            messages, contracts.TextContract(), tool_definitions
+        )
+
+    async def _one_completion(
+        self, messages: list[dict], contract, tool_definitions: list[dict] | None
+    ) -> dict:
+        if self.calls_used >= (self.max_calls or 0):
+            raise BudgetExceeded(self.calls_used, self.max_calls or 0)
+        self.calls_used += 1
+        response_format = contract.response_format()
+        try:
+            message, usage = await openrouter.complete_message(
+                self.model, messages,
+                tools=tool_definitions, response_format=response_format,
+            )
+        # Specific subclasses before the shared parent: a routine 429 must not
+        # flatten into a generic outage, and the user needs the retry_after.
+        except openrouter.RateLimitedError as exc:
+            raise StepFailure(
+                "rate_limited",
+                "The model provider rate limited this run.",
+                exc.retry_after,
+            ) from exc
+        except openrouter.ModelGoneError as exc:
+            raise StepFailure(
+                "model_gone", f"The model {exc.model} is no longer available."
+            ) from exc
+        except openrouter.UpstreamError as exc:
+            raise StepFailure("upstream_error", str(exc)) from exc
+        if self._open is not None:
+            self._open.model = self.model
+            self._open.tokens_in += int(usage.get("prompt_tokens") or 0)
+            self._open.tokens_out += int(usage.get("completion_tokens") or 0)
+        return message
+
+    async def call_tool(self, name: str, arguments: dict) -> dict:
+        """Dispatch a tool, recording it as its own step row.
+
+        A tool call is a step so that every event the stream emits is
+        reconstructible from the database — which is what makes replay-then-
+        follow identical to a first connect.
+        """
+        parent = self._open
+        step_id = runs.add_step(self.run_id, f"tool:{name}")
+        ordinal = len(runs.list_steps(self.run_id))
+        started = time.monotonic()
+        self.emit({"type": "step-start", "name": f"tool:{name}", "ordinal": ordinal})
+        result = await self.registry.dispatch(name, arguments, owner=self.owner)
+        latency = int((time.monotonic() - started) * 1000)
+        failed = not result.get("ok")
+        error = result.get("error", {}).get("code") if failed else None
+        runs.finish_step(
+            step_id,
+            status="failed" if failed else "succeeded",
+            latency_ms=latency,
+            error=error,
+        )
+        self.emit({
+            "type": "step-done", "name": f"tool:{name}", "ordinal": ordinal,
+            "status": "failed" if failed else "succeeded",
+            "latency_ms": latency, "error": error,
+        })
+        # A tool failure is something the model sees and continues past, so it
+        # is returned, never raised. Restore the enclosing step for metrics.
+        self._open = parent
+        return result
+
+
+class PipelineScheduler:
+    """Walks a declared step list. The most deterministic scheduler available:
+    call count and order are knowable before the run starts.
+
+    v0.7.0's ceiling arm pins this scheduler — a ceiling that drifts run to run
+    makes the three-arm comparison unreadable.
+    """
+
+    name = "pipeline"
+
+    def __init__(self, steps: list) -> None:
+        self.steps = steps
+
+    async def run(self, ctx: RunContext) -> dict:
+        output: dict = {}
+        for step in self.steps:
+            result = await step(ctx)
+            if isinstance(result, dict):
+                output = result
+        return output
+
+
+def _tool_call_of(message: dict) -> dict | None:
+    """First well-formed tool call in a model reply, or None.
+
+    Model output, so every level may be the wrong shape — never raise while
+    inspecting one (same rule as chat's `_is_web_search_call`).
+    """
+    calls = message.get("tool_calls") if isinstance(message, dict) else None
+    for call in calls or []:
+        if not isinstance(call, dict):
+            continue
+        function = call.get("function")
+        if isinstance(function, dict) and function.get("name"):
+            return call
+    return None
+
+
+class AgentScheduler:
+    """Loops with tools until a terminal tool or an iteration cap.
+
+    Ships as substrate in v0.6.0 with no skill declaring it: the loop gets
+    proven before any user-facing skill's quality depends on it.
+    """
+
+    name = "agent"
+
+    def __init__(
+        self,
+        system: str,
+        tool_names: list[str],
+        terminal_tool: str,
+        max_iterations: int | None = None,
+    ) -> None:
+        self.system = system
+        self.tool_names = tool_names
+        self.terminal_tool = terminal_tool
+        self.max_iterations = max_iterations
+
+    async def run(self, ctx: RunContext) -> dict:
+        cap = self.max_iterations or get_settings().skill_agent_max_iterations
+        definitions = [
+            d for d in ctx.registry.definitions(owner=ctx.owner)
+            if d["function"]["name"] in self.tool_names
+        ]
+        messages: list[dict] = [
+            {"role": "system", "content": self.system},
+            {"role": "user", "content": json.dumps(ctx.inputs)},
+        ]
+        for iteration in range(cap):
+            async with ctx.step(f"agent:{iteration + 1}"):
+                message = await ctx.call_model_raw(
+                    messages, tool_definitions=definitions
+                )
+                call = _tool_call_of(message)
+                if call is None:
+                    text = (message.get("content") or "").strip()
+                    # No tool call AND real text is the agent answering
+                    # directly — a legitimate end. But a reply that tried to
+                    # call a tool and emitted a malformed payload, or returned
+                    # nothing at all, is not an answer: saying so and letting
+                    # it retry beats returning an empty result that reads as
+                    # success.
+                    if text and not message.get("tool_calls"):
+                        return {"text": text}
+                    messages = messages + [
+                        {"role": "assistant", "content": message.get("content")},
+                        {"role": "user", "content": (
+                            "That reply contained no usable tool call. Either "
+                            "call one of the available tools with valid JSON "
+                            "arguments, or answer directly in plain text."
+                        )},
+                    ]
+                    continue
+                name = call["function"]["name"]
+                try:
+                    arguments = json.loads(call["function"].get("arguments") or "{}")
+                except json.JSONDecodeError:
+                    arguments = {}
+                if not isinstance(arguments, dict):
+                    arguments = {}
+                result = await ctx.call_tool(name, arguments)
+                if name == self.terminal_tool and result.get("ok"):
+                    return result.get("data") or {}
+                messages = messages + [
+                    {"role": "assistant", "content": None, "tool_calls": [call]},
+                    {
+                        "role": "tool",
+                        "tool_call_id": call.get("id", ""),
+                        # Tool results are data, not instructions.
+                        "content": json.dumps(result),
+                    },
+                ]
+        raise StepFailure(
+            "iteration_cap", f"The agent did not finish within {cap} iterations."
+        )

app/harness/runs.py

@@ -272,3 +272,27 @@ def sweep_orphans() -> int:
                WHERE status = 'running'"""
         )
     return swept
+
+
+def cancel_run(run_id: int) -> None:
+    """Terminal, and deliberately distinct from 'failed': the user stopped this
+    on purpose, and run history should say so rather than implying a defect.
+
+    Only an active run is touched — cancelling an already-terminal run must not
+    rewrite a finished record.
+    """
+    with _connect() as conn:
+        conn.execute(
+            f"""UPDATE skill_runs
+                SET status = 'cancelled',
+                    error_message = 'The run was cancelled.',
+                    finished_at = datetime('now')
+                WHERE id = ? AND status IN ({','.join('?' * len(_ACTIVE))})""",
+            (run_id, *_ACTIVE),
+        )
+        conn.execute(
+            """UPDATE skill_run_steps
+               SET status = 'failed', error = 'cancelled'
+               WHERE run_id = ? AND status = 'running'""",
+            (run_id,),
+        )

app/services/openrouter.py

@@ -72,10 +72,10 @@ async def stream_chat(
                     yield delta
 
 
-async def _post_completion(payload: dict) -> dict:
-    """POST a chat-completion payload to OpenRouter and return the raw assistant
-    message dict. Owns the headers, timeout, and the full status-code ladder shared
-    by `complete` and `complete_with_tools`."""
+async def _post_completion_body(payload: dict) -> dict:
+    """POST a chat-completion payload to OpenRouter and return the full decoded
+    body. Owns the headers, timeout, and the status-code ladder shared by every
+    caller."""
     settings = get_settings()
     headers = {
         "Authorization": f"Bearer {settings.openrouter_api_key}",
@@ -101,11 +101,56 @@ async def _post_completion(payload: dict) -> dict:
     try:
         # A 200 carrying a non-JSON body is just as malformed as a missing key,
         # and must not escape as a raw decode error mid-stream.
-        return resp.json()["choices"][0]["message"]
-    except (ValueError, KeyError, IndexError, TypeError) as exc:
+        body = resp.json()
+    except ValueError as exc:
+        raise UpstreamError("malformed completion response") from exc
+    if not isinstance(body, dict):
+        raise UpstreamError("malformed completion response")
+    return body
+
+
+def _message_of(body: dict) -> dict:
+    try:
+        return body["choices"][0]["message"]
+    except (KeyError, IndexError, TypeError) as exc:
         raise UpstreamError("malformed completion response") from exc
 
 
+async def _post_completion(payload: dict) -> dict:
+    """The assistant message alone — what `complete`/`complete_with_tools` need."""
+    return _message_of(await _post_completion_body(payload))
+
+
+async def complete_message(
+    model: str | None,
+    messages: list[dict],
+    *,
+    tools: list[dict] | None = None,
+    response_format: dict | None = None,
+) -> tuple[dict, dict]:
+    """One completion, returning the assistant message AND the usage block.
+
+    The harness records tokens per step, so discarding usage the way `complete`
+    does would leave every step row half empty.
+    """
+    settings = get_settings()
+    payload: dict = {
+        "model": model or settings.default_model,
+        "messages": messages,
+        "stream": False,
+    }
+    # Omitted rather than sent as null: some providers 400 on a null
+    # response_format instead of ignoring it.
+    if tools:
+        payload["tools"] = tools
+        payload["tool_choice"] = "auto"
+    if response_format:
+        payload["response_format"] = response_format
+    body = await _post_completion_body(payload)
+    usage = body.get("usage")
+    return _message_of(body), usage if isinstance(usage, dict) else {}
+
+
 async def complete(model: str | None, messages: list[dict[str, str]]) -> str:
     """Run a single non-streaming OpenRouter chat completion, returning the text."""
     settings = get_settings()

app/skills/__init__.py

@@ -0,0 +1,18 @@
+"""Skills: what the harness runs. Control flow is Python; prompt bodies are
+markdown loaded at import."""
+
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from pydantic import BaseModel
+
+
+@dataclass(frozen=True)
+class Skill:
+    name: str
+    label: str
+    description: str
+    input_model: type[BaseModel]
+    scheduler: Any
+    estimated_calls: Callable[[dict], int]
+    owner_only: bool = True