Divide & Conquer: the Scissari executor_run Stall

A GoF-pattern redesign that replaces the blind 14-call retry loop with a testable, classified recovery pipeline.

Target agent: agent-5955b0c2-7922-4ffe-9e43-b116053b80fa (Scissari)  ·  Executor service: Python/uvicorn  ·  Date: 2026-06-05  ·  Updated: 2026-06-06

Scissari timed out waiting for a response — a tool call took too long.
Please try again, or send /new to start a fresh conversation.

🚨 I stalled and reset my conversation — I could NOT finish your request.
• Reason: Repeated 'executor_run' calls hit the 14-call cap
          (the tool is most likely failing and being retried)
• No specific tool error captured — likely a genuine repetition/stall.

1. Why it’s un-fixable today

The current path is one fused blob. executor_run can fail for at least six structurally different reasons (documented across the debugging-executor-run skill and src/tools/toolset.ts), yet they all funnel into the same blind retry() until the cap:

I ran into an issue completing that request — the response was lost
during a tool workflow. Please try again.

Notice that not one of the six correct responses is “retry the same call again.” Yet that is the only thing the loop does. The cap of 14 is just how long it takes to give up.

Current (broken) control flow

2. The Divide & Conquer plan

Split the monolith into small objects, each behind a Pydantic interface, each independently unit-testable. Program to the interface; inject implementations. The loop guard stops being a counter and becomes an explicit state machine fed by a classifier that picks a strategy.

Target control flow

What each split buys us

• The classifier is pure — given an ExecutorFailure it returns a FailureClassification. No network, no state. Trivially unit-tested with the six fixtures above.
• The loop guard is a state machine — its transitions are a table you can assert against. “14” becomes a per-kind budget, not a global blunt counter.
• The circuit breaker stops the 14× spin on a dead executor in 1–2 calls.
• The Memento means a “reset” snapshots state for forensics instead of vaporizing it — the alert can finally say which failure it was.

3. GoF hat — patterns applied

Command Adapter Chain of Responsibility Strategy Factory Method State Memento Observer Facade Decorator Template Method Circuit Breaker*

*Circuit Breaker is a stability pattern (Nygard), included because it directly kills the 14× spin.

4. Pydantic interfaces — stubbed, ready for red tests

All stubs raise NotImplementedError so the first unit-test run is red by construction (TDD). Implementations get filled one interface at a time until green. Drop these into scissari_executor/ on the executor (Python) side.

4.1 Domain models & enums

scissari_executor/models.pyfrom __future__ import annotations

from enum import Enum
from typing import Any, Optional

from pydantic import BaseModel, Field


class FailureKind(str, Enum):
    """The six structurally-distinct ways executor_run fails (F1–F6)."""
    ALLOWLIST_BLOCKED = "allowlist_blocked"      # F1  HTTP 400
    SERVER_RELOAD_500 = "server_reload_500"      # F2  HTTP 500 watchfiles loop
    REQUEST_TIMEOUT = "request_timeout"          # F3  HTTP 408
    EXECUTOR_DOWN = "executor_down"              # F4  ECONNREFUSED / no process
    END_TURN_NO_RETURN = "end_turn_no_return"    # F5  tool_call then end_turn, no tool_return
    PEER_TOOL_RULE_HANG = "peer_tool_rule_hang"  # F6  peer agent max_steps from bad tool_rule
    UNKNOWN = "unknown"                          # never silently retried — always aborts


class RecoveryAction(str, Enum):
    RETRY = "retry"                  # safe, after backoff
    NARROW = "narrow"                # retry only with a tightened command
    FALLBACK = "fallback"            # run client-side (F5)
    ABORT = "abort"                  # stop now; retrying cannot help
    CIRCUIT_OPEN = "circuit_open"    # service dead; stop and alert ops


class TurnState(str, Enum):
    RUNNING = "running"
    RECOVERING = "recovering"
    RESOLVED = "resolved"
    TRIPPED = "tripped"              # replaces the old "reset at 14"


class ExecutorCommand(BaseModel):
    """Command pattern — a reified executor_run invocation."""
    cmd: str
    cwd: Optional[str] = None
    timeout_s: float = 60.0
    allowlist_key: Optional[str] = None

    def fingerprint(self) -> str:
        """Stable hash for de-dup / repetition detection. STUB."""
        raise NotImplementedError


class ExecutorResponse(BaseModel):
    ok: bool
    status: int
    stdout: str = ""
    stderr: str = ""
    duration_s: float = 0.0


class ExecutorFailure(BaseModel):
    """What IExecutorClient raises/returns on any non-success."""
    status: Optional[int] = None          # HTTP status if any
    transport_error: Optional[str] = None  # e.g. "ECONNREFUSED"
    detail: str = ""                       # server 'detail' body
    raw: dict[str, Any] = Field(default_factory=dict)


class FailureClassification(BaseModel):
    kind: FailureKind
    retryable: bool
    recommended_action: RecoveryAction
    evidence: str                          # human-readable WHY (fixes the "no error captured" gap)
    classifier_name: str


class RecoveryOutcome(BaseModel):
    action: RecoveryAction
    backoff_ms: int = 0
    next_command: Optional[ExecutorCommand] = None  # set when NARROW/FALLBACK
    reason: str = ""


class GuardVerdict(BaseModel):
    state: TurnState
    should_continue: bool
    calls_used: int
    budget_for_kind: int
    snapshot_id: Optional[str] = None      # set when state == TRIPPED


class StallReport(BaseModel):
    """The Observer payload — finally carries a concrete reason."""
    agent_id: str
    classification: Optional[FailureClassification]
    calls_used: int
    final_state: TurnState
    snapshot_id: Optional[str]
    message: str

4.2 Ports / interfaces (ABCs — program to these)

scissari_executor/interfaces.pyfrom __future__ import annotations

from abc import ABC, abstractmethod
from typing import Optional, Sequence

from .models import (
    ExecutorCommand, ExecutorResponse, ExecutorFailure,
    FailureClassification, RecoveryOutcome, GuardVerdict,
    StallReport, FailureKind,
)


class IExecutorClient(ABC):
    """Adapter over the HTTP executor service (uvicorn @ 127.0.0.1:8787)."""
    @abstractmethod
    async def run(self, cmd: ExecutorCommand) -> ExecutorResponse:
        """Return ExecutorResponse on success; raise ExecutorFailureError otherwise. STUB."""
        raise NotImplementedError


class IFailureClassifier(ABC):
    """One link in the Chain of Responsibility. Returns None to defer to the next link."""
    name: str
    @abstractmethod
    def classify(self, failure: ExecutorFailure) -> Optional[FailureClassification]:
        raise NotImplementedError


class IFailureClassifierChain(ABC):
    @abstractmethod
    def classify(self, failure: ExecutorFailure) -> FailureClassification:
        """First matching link wins; falls back to FailureKind.UNKNOWN. STUB."""
        raise NotImplementedError


class IRecoveryStrategy(ABC):
    """Strategy — decides what to do about one classified failure."""
    handles: FailureKind
    @abstractmethod
    async def recover(
        self, cmd: ExecutorCommand, classification: FailureClassification
    ) -> RecoveryOutcome:
        raise NotImplementedError


class IRecoveryStrategyFactory(ABC):
    @abstractmethod
    def for_kind(self, kind: FailureKind) -> IRecoveryStrategy:
        """Map a FailureKind to its Strategy. STUB."""
        raise NotImplementedError


class ICircuitBreaker(ABC):
    @abstractmethod
    def allow(self) -> bool:
        """False when the circuit is open (executor presumed dead). STUB."""
        raise NotImplementedError
    @abstractmethod
    def record_success(self) -> None:
        raise NotImplementedError
    @abstractmethod
    def record_failure(self, kind: FailureKind) -> None:
        raise NotImplementedError


class ILoopGuard(ABC):
    """State machine that replaces the blind `count < 14` counter."""
    @abstractmethod
    def register(self, cmd: ExecutorCommand, outcome: RecoveryOutcome) -> GuardVerdict:
        """Advance the state machine; enforce per-kind budgets & repetition. STUB."""
        raise NotImplementedError
    @abstractmethod
    def reset(self) -> None:
        raise NotImplementedError


class IConversationSnapshotStore(ABC):
    """Memento store — snapshot before any TRIPPED reset."""
    @abstractmethod
    def capture(self, agent_id: str, transcript: Sequence[dict]) -> str:
        """Persist a snapshot, return its id. STUB."""
        raise NotImplementedError
    @abstractmethod
    def restore(self, snapshot_id: str) -> Sequence[dict]:
        raise NotImplementedError


class IAlertSink(ABC):
    """Observer — Telegram, scissari-alerts.jsonl, dashboard LED all implement this."""
    @abstractmethod
    async def emit(self, report: StallReport) -> None:
        raise NotImplementedError


class IExecutorRunService(ABC):
    """Facade — the ONLY thing Scissari's turn loop calls."""
    @abstractmethod
    async def execute(self, cmd: ExecutorCommand, agent_id: str) -> ExecutorResponse:
        """Orchestrate breaker -> client -> classify -> strategy -> guard -> alert. STUB."""
        raise NotImplementedError

4.3 Template-method strategy base (one concrete skeleton)

scissari_executor/strategies.pyfrom __future__ import annotations

from .interfaces import IRecoveryStrategy
from .models import (
    ExecutorCommand, FailureClassification, RecoveryOutcome,
    RecoveryAction, FailureKind,
)


class BaseRecoveryStrategy(IRecoveryStrategy):
    """Template Method: recover() is fixed; subclasses implement _decide()."""
    async def recover(
        self, cmd: ExecutorCommand, classification: FailureClassification
    ) -> RecoveryOutcome:
        if not classification.retryable:
            return RecoveryOutcome(action=RecoveryAction.ABORT,
                                   reason=classification.evidence)
        return self._decide(cmd, classification)   # subclass hook

    def _decide(self, cmd: ExecutorCommand,
                c: FailureClassification) -> RecoveryOutcome:
        raise NotImplementedError


class AllowlistAbortStrategy(BaseRecoveryStrategy):       # F1
    handles = FailureKind.ALLOWLIST_BLOCKED
    def _decide(self, cmd, c) -> RecoveryOutcome:
        raise NotImplementedError

class BackoffRetryStrategy(BaseRecoveryStrategy):         # F2
    handles = FailureKind.SERVER_RELOAD_500
    def _decide(self, cmd, c) -> RecoveryOutcome:
        raise NotImplementedError

class NarrowCommandStrategy(BaseRecoveryStrategy):        # F3
    handles = FailureKind.REQUEST_TIMEOUT
    def _decide(self, cmd, c) -> RecoveryOutcome:
        raise NotImplementedError

class ClientSideFallbackStrategy(BaseRecoveryStrategy):   # F5
    handles = FailureKind.END_TURN_NO_RETURN
    def _decide(self, cmd, c) -> RecoveryOutcome:
        raise NotImplementedError

5. The failing unit tests (red phase)

Each interface gets a focused test. Because every stub raises NotImplementedError, this suite is red on first run — exactly what we want before implementing. The classifier fixtures are the six real failure bodies pulled from the executor logs / chunk logs.

tests/test_classifier.py  ·  pytestimport pytest
from scissari_executor.models import ExecutorFailure, FailureKind, RecoveryAction
from scissari_executor.classifiers import build_default_chain   # to be written


CASES = [
    # (ExecutorFailure fixture, expected kind, expected retryable)
    (ExecutorFailure(status=400, detail="Command not in allowlist: /x/run.sh"),
        FailureKind.ALLOWLIST_BLOCKED, False),
    (ExecutorFailure(status=500, detail="watchfiles reload"),
        FailureKind.SERVER_RELOAD_500, True),
    (ExecutorFailure(status=408, detail="timed out"),
        FailureKind.REQUEST_TIMEOUT, True),
    (ExecutorFailure(transport_error="ECONNREFUSED"),
        FailureKind.EXECUTOR_DOWN, False),
    (ExecutorFailure(detail="end_turn with no tool_return_message"),
        FailureKind.END_TURN_NO_RETURN, True),
    (ExecutorFailure(detail="max_steps: peer required_before_exit send_message"),
        FailureKind.PEER_TOOL_RULE_HANG, False),
]


@pytest.mark.parametrize("failure,kind,retryable", CASES)
def test_chain_classifies_each_failure_mode(failure, kind, retryable):
    chain = build_default_chain()
    result = chain.classify(failure)
    assert result.kind == kind
    assert result.retryable is retryable
    assert result.evidence            # MUST capture a reason (kills "no error captured")


def test_unknown_failure_never_retries():
    chain = build_default_chain()
    result = chain.classify(ExecutorFailure(detail="something nobody mapped"))
    assert result.kind == FailureKind.UNKNOWN
    assert result.recommended_action == RecoveryAction.ABORT

tests/test_loop_guard.py  ·  pytestfrom scissari_executor.guard import LoopGuard          # to be written
from scissari_executor.models import (
    ExecutorCommand, RecoveryOutcome, RecoveryAction, TurnState,
)


def test_identical_command_trips_fast_not_at_14():
    """The bug: 14 identical failing calls. Fix: trip after the per-kind budget (e.g. 2)."""
    guard = LoopGuard(budget_per_kind={"server_reload_500": 2})
    cmd = ExecutorCommand(cmd="ls /huge")
    retry = RecoveryOutcome(action=RecoveryAction.RETRY)
    guard.register(cmd, retry)
    verdict = guard.register(cmd, retry)
    assert verdict.state == TurnState.TRIPPED
    assert verdict.calls_used == 2            # NOT 14
    assert verdict.snapshot_id is not None    # Memento captured before trip


def test_abort_outcome_trips_immediately():
    guard = LoopGuard()
    verdict = guard.register(
        ExecutorCommand(cmd="bad"),
        RecoveryOutcome(action=RecoveryAction.ABORT, reason="allowlist"),
    )
    assert verdict.should_continue is False
    assert verdict.state == TurnState.TRIPPED


def test_successful_recovery_resolves():
    guard = LoopGuard()
    verdict = guard.register(
        ExecutorCommand(cmd="ls"),
        RecoveryOutcome(action=RecoveryAction.RETRY),
    )
    assert verdict.state in (TurnState.RUNNING, TurnState.RECOVERING)

tests/test_circuit_breaker.py  ·  pytestfrom scissari_executor.breaker import CircuitBreaker   # to be written
from scissari_executor.models import FailureKind


def test_opens_after_two_executor_down_and_stops_the_spin():
    cb = CircuitBreaker(threshold=2)
    assert cb.allow() is True
    cb.record_failure(FailureKind.EXECUTOR_DOWN)
    cb.record_failure(FailureKind.EXECUTOR_DOWN)
    assert cb.allow() is False        # would have spun 14× before this fix

tests/test_service_facade.py  ·  pytest (integration, fakes injected)import pytest
from scissari_executor.service import ExecutorRunService   # to be written
from scissari_executor.models import ExecutorCommand, FailureKind
from tests.fakes import FakeExecutorClient, FakeAlertSink


@pytest.mark.asyncio
async def test_dead_executor_aborts_and_alerts_with_reason():
    sink = FakeAlertSink()
    service = ExecutorRunService(
        client=FakeExecutorClient(always_fail="ECONNREFUSED"),
        alert_sink=sink,
        # classifier/factory/guard/breaker injected via DI
    )
    with pytest.raises(Exception):
        await service.execute(ExecutorCommand(cmd="ls"), agent_id="agent-5955...")
    assert sink.last.classification.kind == FailureKind.EXECUTOR_DOWN
    assert "No specific tool error" not in sink.last.message   # the old lie is gone

6. Build order (red → green, one interface at a time)

1. Models — models.py. Implement ExecutorCommand.fingerprint(). → greens its own model test
2. Classifier chain — six small classifiers + build_default_chain(). → greens test_classifier.py
3. Circuit breaker — counting + open/half-open/closed. → greens test_circuit_breaker.py
4. Strategies — fill the four _decide() hooks + factory. → greens strategy tests
5. Loop guard — state machine + per-kind budgets + Memento capture. → greens test_loop_guard.py
6. Alert sinks — Telegram + jsonl + LED Observers behind IAlertSink.
7. Facade — wire it all with DI. → greens test_service_facade.py
8. Swap — replace the old blind loop in lettabot with a call to IExecutorRunService.execute(). Delete the bare count < 14.

7. F7 — the transport/session layer (today’s failure)

F1–F6 all live inside executor_run. F7 is a different concern entirely and the executor scaffold cannot see it: it is lettabot’s session lifecycle around the letta-code SDK subprocess. Divide it out as its own testable unit — do not bolt it onto the classifier.

What actually happened (from the lettabot log)

Two bugs, fused:

• (a) Coarse timer. The 300 000 ms inactivity timer measures stream silence, but a long-but-healthy tool call (a big executor_run) produces no stream events either. A slow success is indistinguishable from a hang, so the bot kills a session that was about to succeed.
• (b) No re-spawn. After the session is closed, the next heartbeat calls session.send() on a transport whose subprocess is gone (pid=undefined). It throws Transport not connected instead of lazily re-spawning. The user sees “a tool call took too long.”

The divide (GoF, same discipline as §2)

Unit tests — written & GREEN ✅

Built in scissari_executor/session/; 13 F7 cases pass alongside the 16 executor cases (pytest -q → 29 passed).

• test_session_health.py — a tool call exceeding stream-idle (300s) but inside its own deadline (900s) is NOT killed; one exceeding its deadline is; genuine idle still trips STREAM_IDLE; tool_return re-arms the idle timer.
• test_resilient_transport.py — send() on a pid=undefined transport triggers exactly one re-spawn then succeeds; repeated spawn failures open the breaker and raise a clean TransportUnavailableError, never the raw Transport not connected.
• test_keepalive.py — inactivity timer is suppressed for the duration of an in-flight tool call and re-armed after tool_return.
• test_session_supervisor.py — facade: a long tool call neither trips nor alerts; the deadline trip emits a StallReport with a concrete reason; a heartbeat send() re-spawns a dead session.

Port: port-typescript/sessionSupervisor.ts is a 1:1 TypeScript drop-in for lettabot (Node). Adoption steps: WIRING.md.

DEPLOYED 2026-06-06: rather than bolt the standalone module onto lettabot, the same F7 semantics were implemented against lettabot’s native session primitives in src/core/bot.ts — bug (a) = inFlightToolCalls + TOOL_CALL_DEADLINE_MS gating resetStreamInactivityTimer(); bug (b) = isTransportDeadError() + a re-spawn/retry branch in runSession() reusing invalidateSession/ensureSessionForKey. Committed 2ae0d25; tsc clean. Restart lettabot to activate.

Cross-references for F7 (lettabot, Win11): dist/core/bot.js (Stream inactivity timeout, Invalidating session, trySend/runSession), dist/cron/heartbeat.js (runHeartbeat → dead-transport send), @letta-ai/letta-code-sdk SubprocessTransport.write (throws Transport not connected).

Cross-references in this repo: src/tools/toolset.ts (findHangingToolRules — F6), src/agent/multi-agent-tool-fallback.ts (CLIENT_SIDE_FALLBACK_TOOLS — F5), src/skills/custom/debugging-executor-run/SKILL.md (F1–F4 fingerprints), src/integration-tests/agent-tool-rule-audit.integration.test.ts (live F6 guard).