Skip to content

stream.executor

stream.executor

Headless executor: drive a coordinator graph to completion via a Driver.

This replaces the human-in-Cursor coordinator loop with deterministic Python. For each wave it claims ready tasks (respecting path-scoped write leases), spawns the agent through a :class:~stream.protocols.Driver, submits the result, and -- on a checker FAIL -- deterministically appends a repair sub-graph.

The coordinator functions are imported lazily and called in-process; they return JSON strings, which we decode here.

Executor dataclass

Source code in stream/executor.py 
 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
@dataclass
class Executor:
    driver: Driver
    concurrency: int = 1
    extension_policy: ExtensionPolicy | None = None
    fix_agent: str = "scribe"
    _registry: list = field(default_factory=zettelkasten_tools)

    # -- coordinator bridge (lazy) -----------------------------------------

    @staticmethod
    def _coord():
        from coordinator import server as coord

        return coord

    # -- public -------------------------------------------------------------

    def run(self, run_id: str = "") -> dict[str, Any]:
        """Drive the graph identified by ``run_id`` until no tasks remain ready."""
        coord = self._coord()
        policy = self.extension_policy or default_extension_policy
        stalls = 0

        while True:
            ready = json.loads(coord.get_ready_tasks(run_id))
            if not ready:
                break

            claimed = self._claim_wave(ready, run_id)
            if not claimed:
                # Everything ready is blocked by a lease whose holder is still
                # running elsewhere. Back off briefly, then retry.
                stalls += 1
                if stalls > 50:
                    raise RuntimeError(
                        "Executor stalled: ready tasks remain but none claimable "
                        "(possible lease deadlock or foreign run holding a lease)."
                    )
                time.sleep(0.2)
                continue
            stalls = 0

            outcomes = self._run_wave(claimed)
            for outcome in outcomes:
                coord.submit_result(
                    outcome.task_id, outcome.result.text, success=outcome.success
                )

            failures = [o for o in outcomes if not o.success]
            if failures:
                self._maybe_extend(failures, policy, run_id)

        return json.loads(coord.get_report(run_id))

    # -- wave mechanics -----------------------------------------------------

    def _claim_wave(self, ready: list[dict], run_id: str) -> list[dict]:
        """Claim as many ready tasks as possible; skip blocked ones."""
        coord = self._coord()
        claimed: list[dict] = []
        for task in ready:
            res = json.loads(coord.claim_task(task["task_id"], run_id))
            if res.get("blocked"):
                continue
            # Merge any claim-time enrichment (model/schema) back onto the task.
            merged = {**task, **{k: v for k, v in res.items() if v is not None}}
            claimed.append(merged)
        return claimed

    def _run_wave(self, claimed: list[dict]) -> list[TaskOutcome]:
        if self.concurrency <= 1 or len(claimed) == 1:
            return [self._run_task(t) for t in claimed]
        with ThreadPoolExecutor(max_workers=self.concurrency) as pool:
            return list(pool.map(self._run_task, claimed))

    def _run_task(self, task: dict) -> TaskOutcome:
        coord = self._coord()
        task_id = task["task_id"]
        agent = task.get("agent", "")

        persona_info = json.loads(coord.get_persona(agent))
        role = persona_info.get("role", "checker")
        persona = persona_info.get("persona", "")

        context = json.loads(coord.get_task_context(task_id))
        prompt = _build_prompt(context, task)
        tools = tools_for_role(role, self._registry)

        result = self.driver.run_agent(
            role=role,
            persona=persona,
            prompt=prompt,
            tools=tools,
            model=task.get("model", ""),
            ctx={"task": task, "schema": task.get("schema")},
        )

        verdict = _parse(result.text, agent)
        success, severity = _verdict_status(verdict)
        return TaskOutcome(
            task=task,
            result=result,
            verdict=verdict,
            success=success,
            severity=severity,
        )

    # -- extensions ---------------------------------------------------------

    def _maybe_extend(
        self, failures: list[TaskOutcome], policy: ExtensionPolicy, run_id: str
    ) -> None:
        new_tasks = policy(failures, self)
        if not new_tasks:
            return
        coord = self._coord()
        after = [f.task_id for f in failures]
        context = _failure_context(failures)
        import logging
        try:
            out = coord.extend_graph(new_tasks, after, context=context, run_id=run_id)
        except Exception as exc:
            # Hard ceiling reached or invalid extension -- stop extending, let
            # the run finish.
            logging.getLogger(__name__).warning(
                "extend_graph rejected (%s); leaving failures in the report.", exc
            )
            return
        # The soft cap returns a structured signal rather than raising. This
        # executor is autonomous (no human to authorize a force=True override),
        # so the correct response is to stop extending and let the run finish.
        try:
            import json as _json

            if isinstance(out, str) and _json.loads(out).get("soft_cap_reached"):
                logging.getLogger(__name__).warning(
                    "extend_graph soft cap reached; leaving failures in the report."
                )
        except Exception:
            pass

run 

run(run_id: str = '') -> dict[str, Any]

Drive the graph identified by run_id until no tasks remain ready.

Source code in stream/executor.py 
 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
def run(self, run_id: str = "") -> dict[str, Any]:
    """Drive the graph identified by ``run_id`` until no tasks remain ready."""
    coord = self._coord()
    policy = self.extension_policy or default_extension_policy
    stalls = 0

    while True:
        ready = json.loads(coord.get_ready_tasks(run_id))
        if not ready:
            break

        claimed = self._claim_wave(ready, run_id)
        if not claimed:
            # Everything ready is blocked by a lease whose holder is still
            # running elsewhere. Back off briefly, then retry.
            stalls += 1
            if stalls > 50:
                raise RuntimeError(
                    "Executor stalled: ready tasks remain but none claimable "
                    "(possible lease deadlock or foreign run holding a lease)."
                )
            time.sleep(0.2)
            continue
        stalls = 0

        outcomes = self._run_wave(claimed)
        for outcome in outcomes:
            coord.submit_result(
                outcome.task_id, outcome.result.text, success=outcome.success
            )

        failures = [o for o in outcomes if not o.success]
        if failures:
            self._maybe_extend(failures, policy, run_id)

    return json.loads(coord.get_report(run_id))

default_extension_policy 

default_extension_policy(failures: list[TaskOutcome], executor: Executor) -> list[dict] | None

Conservative repair: act only on P0 checker failures.

Appends one fix task (the configured implementer) followed by a re-run of each failed checker. P1/P2 and bare CONCERNs are reported, not auto-fixed.

Source code in stream/executor.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def default_extension_policy(
    failures: list[TaskOutcome], executor: Executor
) -> list[dict] | None:
    """Conservative repair: act only on P0 checker failures.

    Appends one fix task (the configured implementer) followed by a re-run of
    each failed checker. P1/P2 and bare CONCERNs are reported, not auto-fixed.
    """
    p0 = [f for f in failures if f.severity == "P0"]
    if not p0:
        return None

    fix_alias = "fix"
    tasks: list[dict] = [
        {
            "alias": fix_alias,
            "agent": executor.fix_agent,
            "description": (
                "Repair the issues found by the failed checker(s). See the "
                "failure context for specific findings."
            ),
            "depends_on": [],
            "writes": [],  # dynamic-mode lease; scribe leases per-box itself
        }
    ]
    for i, f in enumerate(p0):
        tasks.append(
            {
                "alias": f"recheck{i}",
                "agent": f.agent,
                "description": f"Re-verify after the fix: {f.task.get('description', '')}",
                "depends_on": [fix_alias],
            }
        )
    return tasks