Skip to content

Hook entry points and streaming

octomate_cli.emit

Forward a native session's hook payload from stdin to Octomate's hook router.

Both agents' installers write command hooks that run this script — Codex because it has no http hook handler at all, Claude so the settings file stays free of hosts and credentials: with no --url pinned, the router's address is resolved when the hook fires, from OCTOMATE_CLI_URL or the client config file, and switching servers is an environment switch, not a re-install. The transport is HTTP either way: this only carries stdin to the router at <base>{--path}.

Run by absolute path, never as python -m octomate..., and imports nothing from octomate. Both matter: octomate/__init__.py builds Octomate, which pulls in pydantic-ai and costs ~1.9s to import — per hook, on a handler the session blocks on, several times a turn. Codex session names are read through its SDK without loading turn history; the observer never resumes or runs the session.

Keep server-package imports out of this script. The environment variable names, the hook path, and the client-config resolution below are duplicated from octomate_cli/config.py and octomate_cli/tentacles/codex/hooks.py for the same reason — this module cannot import them without paying for a package. Change them together; the tests hold the copies to the canonical ones.

TOKEN_ENV module-attribute 

TOKEN_ENV = 'OCTOMATE_CLI_TOKEN'

OCTOMATE_URL_ENV module-attribute 

OCTOMATE_URL_ENV = 'OCTOMATE_CLI_URL'

CODEX_HOOK_PATH module-attribute 

CODEX_HOOK_PATH = '/hooks/codex'

HOOK_TIMEOUT module-attribute 

HOOK_TIMEOUT = 10

USAGE module-attribute 

USAGE = (
    "usage: emit.py --path <hook-path> [--url <hook-url>]"
)

args module-attribute 

args = sys.argv[1:]

path module-attribute 

path: str | None = None

url module-attribute 

url: str | None = None

flag module-attribute 

flag = args.pop(0)

tables module-attribute 

tables = [
    (file_config(candidate))
    for candidate in (config_files())
]

base module-attribute 

base = resolved('url', OCTOMATE_URL_ENV, tables)

status module-attribute 

status = 1

config_files 

config_files() -> tuple[Path, Path]

Project scope first, then user scope — hooks run with cwd at the directory the session lives in, so ./.octomate/cli.toml is that project's own override.

Source code in cli/octomate_cli/emit.py 
45
46
47
48
49
50
51
def config_files() -> tuple[Path, Path]:
    """Project scope first, then user scope — hooks run with cwd at the directory the
    session lives in, so `./.octomate/cli.toml` is that project's own override."""
    return (
        Path.cwd() / ".octomate" / "cli.toml",
        Path.home() / ".config" / "octomate" / "cli.toml",
    )

file_config 

file_config(path: Path) -> dict[str, object]

One file's table, {} when absent. A file that does not parse is reported and treated as absent: this runs on the blocking path of someone's session, where a traceback helps nobody.

Source code in cli/octomate_cli/emit.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
def file_config(path: Path) -> dict[str, object]:
    """One file's table, `{}` when absent. A file that does not parse is reported and
    treated as absent: this runs on the blocking path of someone's session, where a
    traceback helps nobody."""
    try:
        text = path.read_text()
    except OSError:
        return {}
    try:
        return tomllib.loads(text)
    except tomllib.TOMLDecodeError as error:
        print(f"octomate: ignoring unparseable {path}: {error}", file=sys.stderr)
        return {}

resolved 

resolved(
    key: str, env: str, tables: list[dict[str, object]]
) -> str | None

The client's precedence: the environment, then the config files in scope order.

Source code in cli/octomate_cli/emit.py 
69
70
71
72
73
74
75
76
77
78
79
def resolved(key: str, env: str, tables: list[dict[str, object]]) -> str | None:
    """The client's precedence: the environment, then the config files in scope
    order."""
    value = os.environ.get(env)
    if value:
        return value
    for table in tables:
        from_file = table.get(key)
        if isinstance(from_file, str) and from_file:
            return from_file
    return None

main 

main(url: str, token: str | None) -> int
Source code in cli/octomate_cli/emit.py 
 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
def main(url: str, token: str | None) -> int:
    payload = NativeHookEvent.model_validate_json(sys.stdin.read())
    if not token:
        print(
            f"octomate: no credential — {TOKEN_ENV} is unset and the "
            "client config holds none, so this session is not being ingested. "
            "Run `octomate configure --token <api-token>`.",
            file=sys.stderr,
        )
        return 1

    if (
        urlsplit(url).path == CODEX_HOOK_PATH
        and payload.agent_id is None
        and payload.hook_event_name in {"SessionStart", "UserPromptSubmit", "Stop"}
    ):
        try:
            with CodexClient() as client:
                client.initialize()
                name = client.thread_read(
                    payload.session_id, include_turns=False
                ).thread.name
            if name and name.strip():
                payload.session_name = name
        except (CodexError, OSError) as error:
            # An optional title must not prevent delivery of the observed hook.
            print(
                f"octomate: session name lookup for {payload.session_id} failed: {error}",
                file=sys.stderr,
            )

    request = urllib.request.Request(
        url,
        data=payload.model_dump_json(exclude_unset=True).encode(),
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=HOOK_TIMEOUT) as response:
            answer = response.read().decode()
    except (urllib.error.URLError, OSError) as error:
        # Report and get out of the way. A session is a person's own work, and Octomate
        # only observes it; an ingest that cannot deliver must not take the turn down
        # with it.
        print(f"octomate: hook delivery to {url} failed: {error}", file=sys.stderr)
        return 1
    if (
        payload.hook_event_name == "PreToolUse"
        and urlsplit(url).path != CODEX_HOOK_PATH
    ):
        # Claude reads a PreToolUse hook's stdout as its decision: the router's answer
        # is the call's input with its session stamped in.
        print(answer)
    return 0

octomate_cli.launch

Spawn the transcript tail for a native session — the launcher hook.

The forwarding hooks carry the ledger to Octomate over HTTP, but they can start nothing on this machine; the transcript stream needs a local process, and this command hook is what starts it. It fires on every prompt, spawns octomate <agent> tail detached, and returns at once so the turn never waits — the tail itself refuses to run twice per session, so the repeated fire is the liveness check, not a leak. On a Codex SubagentStop, the event's agent_transcript_path rides along as --agent-path: the running tail cannot see which sibling rollout is a child of its session, so the hook that knows hands it the path (the tail spools it and exits when one already holds the session).

The stream address is pinned only when the install pinned --url; otherwise it is derived when the hook fires, from OCTOMATE_CLI_URL or the client config file — the same resolution the forwarding hooks follow. With none of them, nothing spawns and nothing is said: the emit hook on the same event already complained on stderr.

Run by absolute path, never as python -m octomate..., and imports nothing from octomate: octomate/__init__.py builds Octomate, which costs ~1.9s to import, and this command is on the blocking path of every prompt. Shared hook schemas live in the lightweight protocol package. The environment variable name and the client-config resolution are duplicated from octomate_cli/config.py for the same reason — change them together.

Prints nothing on success: a UserPromptSubmit hook's stdout is injected into the turn's context, so silence is the only correct answer.

OCTOMATE_URL_ENV module-attribute 

OCTOMATE_URL_ENV = 'OCTOMATE_CLI_URL'

USAGE module-attribute 

USAGE = "usage: launch.py [--url <stream-url>] [--path <hook-path>] [--agent <claude|codex>] --octomate <bin>"

args module-attribute 

args = sys.argv[1:]

url module-attribute 

url: str | None = None

path module-attribute 

path: str | None = None

agent module-attribute 

agent = 'claude'

octomate_bin module-attribute 

octomate_bin: str | None = None

flag module-attribute 

flag = args.pop(0)

config_files 

config_files() -> tuple[Path, Path]

Project scope first, then user scope — mirroring emit.py's resolution.

Source code in cli/octomate_cli/launch.py 
47
48
49
50
51
52
def config_files() -> tuple[Path, Path]:
    """Project scope first, then user scope — mirroring `emit.py`'s resolution."""
    return (
        Path.cwd() / ".octomate" / "cli.toml",
        Path.home() / ".config" / "octomate" / "cli.toml",
    )

configured_base 

configured_base() -> str | None

The server base from the environment, then the config files in scope order. A file that does not parse counts as absent — the emit hook on the same event reports it.

Source code in cli/octomate_cli/launch.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def configured_base() -> str | None:
    """The server base from the environment, then the config files in scope order. A
    file that does not parse counts as absent — the emit hook on the same event
    reports it."""
    value = os.environ.get(OCTOMATE_URL_ENV)
    if value:
        return value
    for path in config_files():
        try:
            table = tomllib.loads(path.read_text())
        except (OSError, tomllib.TOMLDecodeError):
            continue
        url = table.get("url")
        if isinstance(url, str) and url:
            return url
    return None

stream_url 

stream_url(url: str | None, path: str | None) -> str | None

The pinned stream URL, or one derived from the resolved base — the same http(s) → ws(s) + /stream derivation the installer's stream_url_for does, duplicated here because this script cannot import the package.

Source code in cli/octomate_cli/launch.py 
73
74
75
76
77
78
79
80
81
82
83
def stream_url(url: str | None, path: str | None) -> str | None:
    """The pinned stream URL, or one derived from the resolved base — the same
    `http(s) → ws(s)` + `/stream` derivation the installer's `stream_url_for` does,
    duplicated here because this script cannot import the package."""
    if url is not None:
        return url
    base = configured_base()
    if base is None or path is None:
        return None
    scheme, _, rest = base.rstrip("/").partition("://")
    return f"{'wss' if scheme == 'https' else 'ws'}://{rest}{path}/stream"

main 

main(
    url: str | None,
    path: str | None,
    agent: str,
    octomate_bin: str,
) -> int
Source code in cli/octomate_cli/launch.py 
 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
def main(url: str | None, path: str | None, agent: str, octomate_bin: str) -> int:
    event = NativeHookEvent.model_validate_json(sys.stdin.read())
    if event.transcript_path is None:
        # An event that names no transcript has nothing to tail; the forwarding hooks
        # still carry the ledger, so this is not the place to complain.
        return 0
    target = stream_url(url, path)
    if target is None:
        return 0
    command = [
        octomate_bin,
        agent,
        "tail",
        "--session",
        event.session_id,
        "--path",
        event.transcript_path,
        "--url",
        target,
    ]
    if event.cwd:
        command += ["--cwd", event.cwd]
    if event.agent_transcript_path:
        command += ["--agent-path", event.agent_transcript_path]
    subprocess.Popen(
        command,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        # Detached: the tail outlives this hook and never holds the turn.
        start_new_session=True,
    )
    return 0

octomate_cli.streaming.files

octomate claude tail / octomate codex tail — the client half of the transcript stream.

Runs on the machine a native session lives on: watches its transcript (and the subagent files beside it), frames complete lines, and ships them raw to Octomate's stream endpoint. No parsing happens here — the server's tailer assembles the turns — so a transcript format change never needs this client updated.

The agents differ only in where a subagent's file lives. Claude keeps them under <session-id>/subagents/, discoverable by glob and keyed by agent id. Codex writes a child rollout as a sibling file only its content identifies, so the launcher hook spools the exact path each SubagentStop names into a per-session file here, and the tail ships every spooled file keyed by its basename — the server classifies each from its own opening session_meta line, keeping even that parsing server-side.

Spawned per session by the launcher hook (launch.py), detached so the turn never waits on it; one instance per config scope, agent and session, enforced with a file lock the OS releases with the process. The server owns every cursor: each connect re-asks where to resume, so this process holds no durable state and a crash anywhere costs a re-stream, never a loss. It ends when the server says finalize (the turn stopped or the session ended; the next prompt's launcher spawns a fresh tail), when the session goes quiet past the idle window (drain and eof — the same bet the server tailer's idle reclaim makes), or when the server refuses it outright (close code 1008).

IDLE_TIMEOUT module-attribute 

IDLE_TIMEOUT = 5 * 60.0

IDLE_POLL_MS module-attribute 

IDLE_POLL_MS = 1000

BACKOFF_CAP module-attribute 

BACKOFF_CAP = 60.0

REFUSED module-attribute 

REFUSED = 1008

LOCK_GRACE module-attribute 

LOCK_GRACE = 15.0

LOCK_POLL module-attribute 

LOCK_POLL = 0.2

SNAPSHOT_CHUNK_SIZE module-attribute 

SNAPSHOT_CHUNK_SIZE = 64 * 1024

SNAPSHOT_ACK_TIMEOUT module-attribute 

SNAPSHOT_ACK_TIMEOUT = 60.0

CLIENT_VERSION module-attribute 

CLIENT_VERSION = version('octomate-cli')

FileCursor dataclass

One file's read state. Complete lines only leave through read_lines; a trailing fragment stays unread and is re-read whole next time, so a line split across two reads is never half-shipped.

Source code in cli/octomate_cli/streaming/files.py 
 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
@dataclass
class FileCursor:
    """One file's read state. Complete lines only leave through `read_lines`; a
    trailing fragment stays unread and is re-read whole next time, so a line split
    across two reads is never half-shipped."""

    path: Path
    offset: int = 0

    def read_lines(self) -> list[tuple[int, int, bytes]]:
        """New complete lines as (start, end, raw) — offsets in this file's own byte
        space, `end` past the newline each line was framed on."""
        try:
            size = self.path.stat().st_size
        except FileNotFoundError:
            return []
        if size < self.offset:  # truncation guard, mirroring the server pump's
            self.offset = 0
        with self.path.open("rb") as handle:
            handle.seek(self.offset)
            chunk = handle.read()
        lines: list[tuple[int, int, bytes]] = []
        for raw in chunk.split(b"\n")[:-1]:
            start = self.offset
            self.offset += len(raw) + 1
            # Blank lines ship too: the server advances its expected offset from
            # every frame, so skipping one here would read as a gap there.
            lines.append((start, self.offset, raw))
        return lines

path instance-attribute 

path: Path

offset class-attribute instance-attribute 

offset: int = 0

read_lines 

read_lines() -> list[tuple[int, int, bytes]]

New complete lines as (start, end, raw) — offsets in this file's own byte space, end past the newline each line was framed on.

Source code in cli/octomate_cli/streaming/files.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def read_lines(self) -> list[tuple[int, int, bytes]]:
    """New complete lines as (start, end, raw) — offsets in this file's own byte
    space, `end` past the newline each line was framed on."""
    try:
        size = self.path.stat().st_size
    except FileNotFoundError:
        return []
    if size < self.offset:  # truncation guard, mirroring the server pump's
        self.offset = 0
    with self.path.open("rb") as handle:
        handle.seek(self.offset)
        chunk = handle.read()
    lines: list[tuple[int, int, bytes]] = []
    for raw in chunk.split(b"\n")[:-1]:
        start = self.offset
        self.offset += len(raw) + 1
        # Blank lines ship too: the server advances its expected offset from
        # every frame, so skipping one here would read as a gap there.
        lines.append((start, self.offset, raw))
    return lines

SessionTail dataclass

One session's cursors: the transcript's own, and one per subagent file, each seeded from the offset the server's welcome named for it.

Subagent files come from the agent's own layout. With no spool (Claude), they appear under <session-id>/subagents/ and key by agent id. With one (Codex), they are the sibling rollouts the spool names, keyed by basename — a label in this machine's namespace the server classifies by content.

Source code in cli/octomate_cli/streaming/files.py 
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
189
190
191
192
193
194
195
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
@dataclass
class SessionTail:
    """One session's cursors: the transcript's own, and one per subagent file, each
    seeded from the offset the server's welcome named for it.

    Subagent files come from the agent's own layout. With no `spool` (Claude), they
    appear under `<session-id>/subagents/` and key by agent id. With one (Codex),
    they are the sibling rollouts the spool names, keyed by basename — a label in
    this machine's namespace the server classifies by content."""

    session_id: str
    transcript_path: Path
    offsets: dict[str, int] = field(default_factory=dict)
    files: dict[str, FileCursor] = field(default_factory=dict)
    last_active: float = field(default_factory=monotonic)
    spool: Path | None = None
    stored_offset: int | None = None  # None disables transcript uploads.
    receipts: asyncio.Queue[StreamSnapshotStored | ConnectionClosed | ValueError] = (
        field(default_factory=asyncio.Queue)
    )
    spooled: dict[str, Path] = field(default_factory=dict)

    @property
    def subagents_dir(self) -> Path:
        return self.transcript_path.with_suffix("") / "subagents"

    def discover(self) -> list[str]:
        """The file keys present right now: the session itself, plus every subagent
        file — new children join the pump the moment they exist (or are spooled)."""
        keys = [SESSION_FILE]
        if self.spool is not None:
            try:
                lines = self.spool.read_text().splitlines()
            except OSError:
                lines = []
            for entry in lines:
                if entry:
                    path = Path(entry)
                    self.spooled[path.name] = path
            keys += sorted(self.spooled)
        elif self.subagents_dir.is_dir():
            keys += sorted(
                path.stem.removeprefix("agent-")
                for path in self.subagents_dir.glob("agent-*.jsonl")
            )
        return keys

    def cursor(self, key: str) -> FileCursor:
        cursor = self.files.get(key)
        if cursor is None:
            if key == SESSION_FILE:
                path = self.transcript_path
            elif self.spool is not None:
                path = self.spooled[key]
            else:
                path = self.subagents_dir / f"agent-{key}.jsonl"
            cursor = FileCursor(path, self.offsets.get(key, 0))
            self.files[key] = cursor
        return cursor

    async def pump(self, websocket: ClientConnection) -> bool:
        """Ship every complete new line of every file; True if anything went out."""
        sent = False
        keys = self.discover()
        # Interleave files without reordering any file's lines. A parent's backfill
        # must not sit ahead of every live child on the shared connection.
        batches = zip_longest(*(self.cursor(key).read_lines() for key in keys))
        for batch in batches:
            for key, line in zip(keys, batch, strict=True):
                if line is None:
                    continue
                start, end, raw = line
                await websocket.send(
                    StreamLine(
                        agent_id=key or None,
                        start=start,
                        end=end,
                        line=raw.decode("utf-8", errors="replace"),
                    ).model_dump_json()
                )
                sent = True
        if sent:
            self.last_active = monotonic()
        if self.stored_offset is not None:
            await self.upload_snapshot(websocket, start=self.stored_offset)
        return sent

    def snapshot(self, end: int, *, start: int = 0) -> Iterator[bytes]:
        """Read an already streamed range without resending the stored prefix."""
        if start < 0 or end <= start or end > self.cursor(SESSION_FILE).offset:
            raise ValueError("Snapshot boundary must be within the streamed transcript")
        with self.transcript_path.open("rb") as handle:
            if start:
                handle.seek(start - 1)
                if handle.read(1) != b"\n":
                    raise ValueError(
                        "Snapshot boundary must start after a complete line"
                    )
            handle.seek(end - 1)
            if handle.read(1) != b"\n":
                raise ValueError(
                    "Snapshot boundary must end a complete transcript line"
                )
            handle.seek(start)
            remaining = end - start
            while remaining:
                chunk = handle.read(min(remaining, SNAPSHOT_CHUNK_SIZE))
                if not chunk:
                    raise ValueError("Transcript was truncated during snapshot upload")
                remaining -= len(chunk)
                yield chunk

    async def upload_snapshot(self, websocket: ClientConnection, *, start: int) -> None:
        """Upload only new transcript bytes and require their acknowledgment."""
        end = self.cursor(SESSION_FILE).offset
        if start == end:
            return
        if start > end:
            raise ValueError("Transcript was truncated since its last upload")
        request = StreamSnapshotStart(transfer_id=uuid7(), start=start, end=end)
        await websocket.send(request.model_dump_json())
        for chunk in self.snapshot(request.end, start=request.start):
            await websocket.send(chunk)
        async with asyncio.timeout(SNAPSHOT_ACK_TIMEOUT):
            stored = await self.receipts.get()
        if isinstance(stored, (ConnectionClosed, ValueError)):
            raise stored
        if stored.transfer_id != request.transfer_id:
            raise ValueError("Expected acknowledgment for the requested snapshot")
        self.stored_offset = end

session_id instance-attribute 

session_id: str

transcript_path instance-attribute 

transcript_path: Path

offsets class-attribute instance-attribute 

offsets: dict[str, int] = field(default_factory=dict)

files class-attribute instance-attribute 

files: dict[str, FileCursor] = field(default_factory=dict)

last_active class-attribute instance-attribute 

last_active: float = field(default_factory=monotonic)

spool class-attribute instance-attribute 

spool: Path | None = None

stored_offset class-attribute instance-attribute 

stored_offset: int | None = None

receipts class-attribute instance-attribute 

receipts: Queue[
    StreamSnapshotStored | ConnectionClosed | ValueError
] = field(default_factory=asyncio.Queue)

spooled class-attribute instance-attribute 

spooled: dict[str, Path] = field(default_factory=dict)

subagents_dir property 

subagents_dir: Path

discover 

discover() -> list[str]

The file keys present right now: the session itself, plus every subagent file — new children join the pump the moment they exist (or are spooled).

Source code in cli/octomate_cli/streaming/files.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def discover(self) -> list[str]:
    """The file keys present right now: the session itself, plus every subagent
    file — new children join the pump the moment they exist (or are spooled)."""
    keys = [SESSION_FILE]
    if self.spool is not None:
        try:
            lines = self.spool.read_text().splitlines()
        except OSError:
            lines = []
        for entry in lines:
            if entry:
                path = Path(entry)
                self.spooled[path.name] = path
        keys += sorted(self.spooled)
    elif self.subagents_dir.is_dir():
        keys += sorted(
            path.stem.removeprefix("agent-")
            for path in self.subagents_dir.glob("agent-*.jsonl")
        )
    return keys

cursor 

cursor(key: str) -> FileCursor
Source code in cli/octomate_cli/streaming/files.py 
188
189
190
191
192
193
194
195
196
197
198
199
def cursor(self, key: str) -> FileCursor:
    cursor = self.files.get(key)
    if cursor is None:
        if key == SESSION_FILE:
            path = self.transcript_path
        elif self.spool is not None:
            path = self.spooled[key]
        else:
            path = self.subagents_dir / f"agent-{key}.jsonl"
        cursor = FileCursor(path, self.offsets.get(key, 0))
        self.files[key] = cursor
    return cursor

pump async 

pump(websocket: ClientConnection) -> bool

Ship every complete new line of every file; True if anything went out.

Source code in cli/octomate_cli/streaming/files.py 
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
async def pump(self, websocket: ClientConnection) -> bool:
    """Ship every complete new line of every file; True if anything went out."""
    sent = False
    keys = self.discover()
    # Interleave files without reordering any file's lines. A parent's backfill
    # must not sit ahead of every live child on the shared connection.
    batches = zip_longest(*(self.cursor(key).read_lines() for key in keys))
    for batch in batches:
        for key, line in zip(keys, batch, strict=True):
            if line is None:
                continue
            start, end, raw = line
            await websocket.send(
                StreamLine(
                    agent_id=key or None,
                    start=start,
                    end=end,
                    line=raw.decode("utf-8", errors="replace"),
                ).model_dump_json()
            )
            sent = True
    if sent:
        self.last_active = monotonic()
    if self.stored_offset is not None:
        await self.upload_snapshot(websocket, start=self.stored_offset)
    return sent

snapshot 

snapshot(end: int, *, start: int = 0) -> Iterator[bytes]

Read an already streamed range without resending the stored prefix.

Source code in cli/octomate_cli/streaming/files.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def snapshot(self, end: int, *, start: int = 0) -> Iterator[bytes]:
    """Read an already streamed range without resending the stored prefix."""
    if start < 0 or end <= start or end > self.cursor(SESSION_FILE).offset:
        raise ValueError("Snapshot boundary must be within the streamed transcript")
    with self.transcript_path.open("rb") as handle:
        if start:
            handle.seek(start - 1)
            if handle.read(1) != b"\n":
                raise ValueError(
                    "Snapshot boundary must start after a complete line"
                )
        handle.seek(end - 1)
        if handle.read(1) != b"\n":
            raise ValueError(
                "Snapshot boundary must end a complete transcript line"
            )
        handle.seek(start)
        remaining = end - start
        while remaining:
            chunk = handle.read(min(remaining, SNAPSHOT_CHUNK_SIZE))
            if not chunk:
                raise ValueError("Transcript was truncated during snapshot upload")
            remaining -= len(chunk)
            yield chunk

upload_snapshot async 

upload_snapshot(
    websocket: ClientConnection, *, start: int
) -> None

Upload only new transcript bytes and require their acknowledgment.

Source code in cli/octomate_cli/streaming/files.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
async def upload_snapshot(self, websocket: ClientConnection, *, start: int) -> None:
    """Upload only new transcript bytes and require their acknowledgment."""
    end = self.cursor(SESSION_FILE).offset
    if start == end:
        return
    if start > end:
        raise ValueError("Transcript was truncated since its last upload")
    request = StreamSnapshotStart(transfer_id=uuid7(), start=start, end=end)
    await websocket.send(request.model_dump_json())
    for chunk in self.snapshot(request.end, start=request.start):
        await websocket.send(chunk)
    async with asyncio.timeout(SNAPSHOT_ACK_TIMEOUT):
        stored = await self.receipts.get()
    if isinstance(stored, (ConnectionClosed, ValueError)):
        raise stored
    if stored.transfer_id != request.transfer_id:
        raise ValueError("Expected acknowledgment for the requested snapshot")
    self.stored_offset = end

tail_path 

tail_path(
    agent: Literal["claude", "codex", "deepseek"],
    session_id: str,
) -> Path

Lock/spool basename scoped by config location, never its URL or contents.

Source code in cli/octomate_cli/streaming/files.py 
130
131
132
133
134
135
136
137
138
def tail_path(agent: Literal["claude", "codex", "deepseek"], session_id: str) -> Path:
    """Lock/spool basename scoped by config location, never its URL or contents."""
    config_path = project_config_path()
    if not config_path.is_file():
        config_path = user_config_path()
    key = json.dumps((str(config_path), agent, session_id)).encode()
    return (
        Path(tempfile.gettempdir()) / f"octomate-tail-{hashlib.sha256(key).hexdigest()}"
    )

stream_session async 

stream_session(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    spool: Path | None = None,
) -> bool

One connection's life. True when the session is done (the server said finalize, or it went idle and drained out); False to reconnect and resume.

Source code in cli/octomate_cli/streaming/files.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
async def stream_session(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    spool: Path | None = None,
) -> bool:
    """One connection's life. True when the session is done (the server said finalize,
    or it went idle and drained out); False to reconnect and resume."""
    async with connect(
        url, additional_headers={"Authorization": f"Bearer {token}"}
    ) as websocket:
        await websocket.send(
            StreamHello(
                protocol=STREAM_PROTOCOL,
                session_id=session_id,
                transcript_path=str(transcript_path),
                cwd=cwd,
                client_version=CLIENT_VERSION,
            ).model_dump_json()
        )
        welcome = server_message_adapter.validate_json(await websocket.recv())
        if not isinstance(welcome, StreamWelcome):
            raise RuntimeError(f"expected a welcome message, got {welcome.type!r}")
        tail = SessionTail(
            session_id, transcript_path, offsets=dict(welcome.offsets), spool=spool
        )
        checkpoint = welcome.transcript
        if checkpoint is not None:
            digest = hashlib.sha256()
            if checkpoint.offset:
                for chunk in tail.snapshot(checkpoint.offset):
                    digest.update(chunk)
            if digest.hexdigest() != checkpoint.sha256:
                raise ValueError("Transcript prefix differs from the stored snapshot")
            tail.stored_offset = checkpoint.offset

        stop = asyncio.Event()
        finalized = asyncio.Event()
        idle = False

        async def receive_server() -> None:
            try:
                while True:
                    message = server_message_adapter.validate_json(
                        await websocket.recv()
                    )
                    if isinstance(message, StreamFinalize):
                        finalized.set()
                        stop.set()
                    elif isinstance(message, StreamSnapshotStored):
                        tail.receipts.put_nowait(message)
                    else:
                        raise ValueError(f"Unexpected server message: {message.type}")
            except (ConnectionClosed, ValueError) as error:
                tail.receipts.put_nowait(error)
            finally:
                stop.set()

        receiver = asyncio.create_task(receive_server())
        try:
            await tail.pump(websocket)
            async for _ in awatch(
                transcript_path.parent,
                stop_event=stop,
                # Subagent files live under `<session-id>/subagents/`, a level down.
                recursive=True,
                rust_timeout=IDLE_POLL_MS,
                yield_on_timeout=True,
            ):
                await tail.pump(websocket)
                if monotonic() - tail.last_active > IDLE_TIMEOUT:
                    idle = True
                    break
            if not finalized.is_set() and not idle:
                return False  # the socket dropped mid-watch: reconnect and resume
            await tail.pump(websocket)  # final drain to EOF
            await websocket.send(StreamEof().model_dump_json())
            # Wait out the server's close so the eof is consumed, bounded so a
            # wedged server cannot park this process forever.
            with contextlib.suppress(ConnectionClosed, TimeoutError):
                async with asyncio.timeout(10):
                    await receiver
            return True
        finally:
            receiver.cancel()
            with contextlib.suppress(ConnectionClosed, asyncio.CancelledError):
                await receiver

run_tail async 

run_tail(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    spool: Path | None = None,
) -> None

The reconnect loop around stream_session: resume after drops, back off while the server is unreachable, and stop retrying once the session itself has gone quiet — a fresh session's launcher starts a fresh tail.

Source code in cli/octomate_cli/streaming/files.py 
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
async def run_tail(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    spool: Path | None = None,
) -> None:
    """The reconnect loop around `stream_session`: resume after drops, back off while
    the server is unreachable, and stop retrying once the session itself has gone
    quiet — a fresh session's launcher starts a fresh tail."""
    attempt = 0
    while True:
        try:
            if await stream_session(
                url, session_id, transcript_path, cwd, token, spool
            ):
                return
            attempt = 0  # the connection worked; the drop was the network's
        except ConnectionClosed as closed:
            code = closed.rcvd.code if closed.rcvd is not None else None
            if code == REFUSED:
                reason = closed.rcvd.reason if closed.rcvd is not None else ""
                print(f"octomate: stream refused: {reason}", file=sys.stderr)
                return
            attempt += 1
        except InvalidStatus as denied:
            # A pre-accept denial arrives as an HTTP handshake rejection, not a
            # close code — `hook_guard`'s 401 — and does not heal by retrying.
            if denied.response.status_code in {401, 403}:
                print(
                    "octomate: stream denied — a valid API token with hooks scope "
                    "is required. Run `octomate configure --token <api-token>`.",
                    file=sys.stderr,
                )
                return
            attempt += 1
        except (OSError, InvalidHandshake, TimeoutError):
            attempt += 1
        try:
            # This single metadata probe stays synchronous in reconnect bookkeeping.
            quiet = time.time() - transcript_path.stat().st_mtime  # noqa: ASYNC240
        except FileNotFoundError:
            return  # the transcript is gone: nothing left to ship
        if quiet > IDLE_TIMEOUT:
            return
        await asyncio.sleep(min(BACKOFF_CAP, float(2**attempt)))

main 

main(
    *,
    agent: Literal["claude", "codex"],
    session_id: str,
    transcript_path: Path,
    url: str,
    cwd: str,
    agent_path: Path | None = None,
) -> None
Source code in cli/octomate_cli/streaming/files.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def main(
    *,
    agent: Literal["claude", "codex"],
    session_id: str,
    transcript_path: Path,
    url: str,
    cwd: str,
    agent_path: Path | None = None,
) -> None:
    runtime_path = tail_path(agent, session_id)
    spool = runtime_path.with_suffix(".paths") if agent == "codex" else None
    # Spool before the lock: when a tail already holds the session, this invocation's
    # whole job is handing it the child's path — the holder re-reads the spool on
    # every pump. O_APPEND keeps concurrent hook fires from clobbering each other.
    if spool is not None and agent_path is not None:
        with spool.open("a") as handle:
            handle.write(f"{agent_path}\n")
    token = cli_settings().token
    if not token:
        print(
            f"octomate: no credential — {CLISettings.env('token')} is unset and the "
            "client config holds none, so this session is not being streamed. "
            "Run `octomate configure --token <api-token>`.",
            file=sys.stderr,
        )
        raise SystemExit(1)
    # One tail per session in this config scope; changing TOML contents does not
    # replace its lock or affect the holder. flock dies with the process.
    lock_path = runtime_path.with_suffix(".lock")
    with lock_path.open("w") as lock:
        deadline = monotonic() + LOCK_GRACE
        while True:
            try:
                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                if monotonic() >= deadline:
                    return
                time.sleep(LOCK_POLL)
        asyncio.run(run_tail(url, session_id, transcript_path, cwd, token, spool))

octomate_cli.streaming.deepseek

octomate deepseek tail — the client half of the dsh event stream.

Runs on the machine a native dsh session lives on, but unlike the Claude and Codex tails it opens no file: dsh's session log is zstd-framed and only advances at checkpoints, while the harness's own /api gateway serves it decoded and unpacked (session/follow and session/page) — cold sessions included, other dsh processes under the same $DSH_HOME included. So this client reads the local gateway and ships each history entry as one framed line to Octomate's stream endpoint, the event's dense seq standing where a file tail's byte offsets stand (start = seq, end = seq + 1). The server assembles the turns; the server never speaks to this machine's dsh.

One safety rule is the only interpretation this client performs beyond framing: for a session whose last turn is still open, the gateway's reader synthesizes turn/end {reason: interrupted} closers in memory — and if the session's own process later completes the turn, the log's real events land on those same seqs with different content. Shipping the synthesis would advance the cursor past seqs whose truth is not yet written. So a fetch whose final event is an interrupted turn/end withholds everything from that turn's turn/start on; the next poll re-reads, and the turn ships once anything follows it — a successor event is the proof the closers are the log's own.

Spawned per session by the launcher hook (launch.py), detached; one instance per config scope, agent and session via the same flock the file tails use. The server owns the cursor: each connect re-asks where to resume (the committed floor), so this process holds no durable state. It ends on the server's finalize (a Stop settled), on the idle window, or on a policy refusal (close 1008, such as a stale protocol).

POLL_INTERVAL module-attribute 

POLL_INTERVAL = 1.0

RPC_TIMEOUT module-attribute 

RPC_TIMEOUT = 10

PAGE_MESSAGES module-attribute 

PAGE_MESSAGES = 200

json_object_adapter module-attribute 

json_object_adapter = TypeAdapter(dict[str, JsonValue])

DshCompatibilityError

Bases: RuntimeError

Authentication or Remote API mismatch requires operator action.

Source code in cli/octomate_cli/streaming/deepseek.py 
95
96
class DshCompatibilityError(RuntimeError):
    """Authentication or Remote API mismatch requires operator action."""

DshHistoryClient

One authenticated cookie session for native history reads.

Source code in cli/octomate_cli/streaming/deepseek.py 
 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
189
190
191
192
193
class DshHistoryClient:
    """One authenticated cookie session for native history reads."""

    url: str
    cookies: CookieJar
    opener: urllib.request.OpenerDirector

    def __init__(self, launch_url: str) -> None:
        parts = urlsplit(launch_url)
        self.url = urlunsplit(parts._replace(path="", query="", fragment=""))
        self.cookies = CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cookies)
        )
        token = parse_qs(parts.query).get(
            "token", [os.environ.get("DSH_LAUNCH_TOKEN")]
        )[0]
        if token:
            query = urlencode({"token": token})
            try:
                with self.opener.open(f"{self.url}/?{query}", timeout=RPC_TIMEOUT):
                    pass
            except (urllib.error.URLError, OSError):
                raise DshCompatibilityError(
                    "Could not authenticate to dsh; save its current launch URL "
                    "with octomate configure --dsh-url, or check DSH_API_URL and DSH_LAUNCH_TOKEN"
                ) from None
            if not self.cookies:
                raise DshCompatibilityError(
                    "dsh did not issue an authentication cookie; save its current "
                    "launch URL with octomate configure --dsh-url, or check DSH_LAUNCH_TOKEN"
                )

    def rpc(self, method: str, args: dict[str, JsonValue]) -> dict[str, JsonValue]:
        body = ClientRequest(rpc_id=str(uuid4()), method=method, payload={"args": args})
        request = urllib.request.Request(
            f"{self.url}/api/{method}",
            data=body.model_dump_json(by_alias=True).encode(),
            headers={"Content-Type": "application/json"},
        )
        try:
            with self.opener.open(request, timeout=RPC_TIMEOUT) as response:
                answer = ServerResponse.model_validate_json(response.read())
        except urllib.error.HTTPError as error:
            raise DshCompatibilityError(
                f"dsh Remote API {method} returned HTTP {error.code}; configure its "
                "launch URL with octomate configure --dsh-url for authentication "
                "and update dsh and Octomate together"
            ) from None
        if isinstance(answer.result, ErrResult):
            raise DshCompatibilityError(
                f"dsh Remote API {method} failed: {answer.result.error.message}"
            )
        return json_object_adapter.validate_python(answer.result.value)

    def snapshot(self, session_id: str) -> dict[str, JsonValue]:
        request = urllib.request.Request(f"{self.url}/api/remote.mux")
        self.cookies.add_cookie_header(request)
        headers = {"Cookie": request.get_header("Cookie", "")}
        parts = urlsplit(self.url)
        url = urlunsplit(
            parts._replace(
                scheme="wss" if parts.scheme == "https" else "ws",
                path="/api/remote.mux",
            )
        )
        with connect_sync(
            url, additional_headers=headers, open_timeout=RPC_TIMEOUT
        ) as socket:
            socket.send(
                RemoteSessionFollow(
                    stream_id=session_id,
                    payload={
                        "args": {
                            "request": {
                                "address": {"kind": "session", "sessionId": session_id},
                                "maxMessages": PAGE_MESSAGES,
                            }
                        }
                    },
                ).model_dump_json(by_alias=True)
            )
            message = remote_message_adapter.validate_json(
                socket.recv(timeout=RPC_TIMEOUT)
            )
            value = message.value if isinstance(message, RemoteItem) else None
            if (
                not isinstance(message, RemoteItem)
                or not isinstance(value, dict)
                or value.get("type") != "snapshot"
            ):
                raise DshCompatibilityError(
                    "dsh session/follow did not return a snapshot; update dsh and Octomate together"
                )
            return value

url instance-attribute 

url: str = urlunsplit(
    parts._replace(path="", query="", fragment="")
)

cookies instance-attribute 

cookies: CookieJar = CookieJar()

opener instance-attribute 

opener: OpenerDirector = urllib.request.build_opener(
    urllib.request.HTTPCookieProcessor(self.cookies)
)

rpc 

rpc(
    method: str, args: dict[str, JsonValue]
) -> dict[str, JsonValue]
Source code in cli/octomate_cli/streaming/deepseek.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def rpc(self, method: str, args: dict[str, JsonValue]) -> dict[str, JsonValue]:
    body = ClientRequest(rpc_id=str(uuid4()), method=method, payload={"args": args})
    request = urllib.request.Request(
        f"{self.url}/api/{method}",
        data=body.model_dump_json(by_alias=True).encode(),
        headers={"Content-Type": "application/json"},
    )
    try:
        with self.opener.open(request, timeout=RPC_TIMEOUT) as response:
            answer = ServerResponse.model_validate_json(response.read())
    except urllib.error.HTTPError as error:
        raise DshCompatibilityError(
            f"dsh Remote API {method} returned HTTP {error.code}; configure its "
            "launch URL with octomate configure --dsh-url for authentication "
            "and update dsh and Octomate together"
        ) from None
    if isinstance(answer.result, ErrResult):
        raise DshCompatibilityError(
            f"dsh Remote API {method} failed: {answer.result.error.message}"
        )
    return json_object_adapter.validate_python(answer.result.value)

snapshot 

snapshot(session_id: str) -> dict[str, JsonValue]
Source code in cli/octomate_cli/streaming/deepseek.py 
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
189
190
191
192
193
def snapshot(self, session_id: str) -> dict[str, JsonValue]:
    request = urllib.request.Request(f"{self.url}/api/remote.mux")
    self.cookies.add_cookie_header(request)
    headers = {"Cookie": request.get_header("Cookie", "")}
    parts = urlsplit(self.url)
    url = urlunsplit(
        parts._replace(
            scheme="wss" if parts.scheme == "https" else "ws",
            path="/api/remote.mux",
        )
    )
    with connect_sync(
        url, additional_headers=headers, open_timeout=RPC_TIMEOUT
    ) as socket:
        socket.send(
            RemoteSessionFollow(
                stream_id=session_id,
                payload={
                    "args": {
                        "request": {
                            "address": {"kind": "session", "sessionId": session_id},
                            "maxMessages": PAGE_MESSAGES,
                        }
                    }
                },
            ).model_dump_json(by_alias=True)
        )
        message = remote_message_adapter.validate_json(
            socket.recv(timeout=RPC_TIMEOUT)
        )
        value = message.value if isinstance(message, RemoteItem) else None
        if (
            not isinstance(message, RemoteItem)
            or not isinstance(value, dict)
            or value.get("type") != "snapshot"
        ):
            raise DshCompatibilityError(
                "dsh session/follow did not return a snapshot; update dsh and Octomate together"
            )
        return value

event_of 

event_of(entry: JsonValue) -> dict[str, JsonValue] | None
Source code in cli/octomate_cli/streaming/deepseek.py 
196
197
198
199
200
def event_of(entry: JsonValue) -> dict[str, JsonValue] | None:
    if not isinstance(entry, dict):
        return None
    event = entry.get("event")
    return event if isinstance(event, dict) else None

seq_of 

seq_of(entry: JsonValue) -> int | None
Source code in cli/octomate_cli/streaming/deepseek.py 
203
204
205
206
207
208
def seq_of(entry: JsonValue) -> int | None:
    event = event_of(entry)
    if event is None:
        return None
    seq = event.get("seq")
    return seq if isinstance(seq, int) else None

new_entries 

new_entries(
    client: DshHistoryClient, session_id: str, cursor: int
) -> list[dict[str, JsonValue]]

The session's history entries at or past cursor, in seq order — pages walked back from the tail until the window reaches the cursor.

Source code in cli/octomate_cli/streaming/deepseek.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def new_entries(
    client: DshHistoryClient, session_id: str, cursor: int
) -> list[dict[str, JsonValue]]:
    """The session's history entries at or past `cursor`, in seq order — pages
    walked back from the tail until the window reaches the cursor."""
    collected: dict[int, dict[str, JsonValue]] = {}
    value = client.snapshot(session_id)
    through_seq = value.get("cursor")
    if not isinstance(through_seq, int):
        raise DshCompatibilityError("dsh snapshot has no cursor")
    while True:
        entries = value.get("records")
        if not isinstance(entries, list):
            raise DshCompatibilityError("dsh history has no records")
        first_seq: int | None = None
        for entry in entries:
            if not isinstance(entry, dict):
                continue
            seq = seq_of(entry)
            if seq is None:
                continue
            collected[seq] = entry
            first_seq = seq if first_seq is None else min(first_seq, seq)
        if not value.get("hasMore") or first_seq is None or first_seq <= cursor:
            break
        value = client.rpc(
            "session/page",
            {
                "request": {
                    "address": {"kind": "session", "sessionId": session_id},
                    "throughSeq": through_seq,
                    "beforeSeq": first_seq,
                    "maxMessages": PAGE_MESSAGES,
                }
            },
        )
    return [collected[seq] for seq in sorted(collected) if seq >= cursor]

reason_kind 

reason_kind(event: dict[str, JsonValue]) -> str | None
Source code in cli/octomate_cli/streaming/deepseek.py 
250
251
252
253
254
255
256
257
258
def reason_kind(event: dict[str, JsonValue]) -> str | None:
    data = event.get("data")
    if not isinstance(data, dict):
        return None
    reason = data.get("reason")
    if not isinstance(reason, dict):
        return None
    kind = reason.get("kind")
    return kind if isinstance(kind, str) else None

shippable 

shippable(
    entries: list[dict[str, JsonValue]],
) -> list[dict[str, JsonValue]]

The prefix safe to ship: everything, unless the fetch ends on an interrupted turn/end — possibly the reader's in-memory synthesis for a turn another process is still writing — in which case the whole trailing turn (its turn/start on) is withheld for a later poll to prove out.

Source code in cli/octomate_cli/streaming/deepseek.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def shippable(entries: list[dict[str, JsonValue]]) -> list[dict[str, JsonValue]]:
    """The prefix safe to ship: everything, unless the fetch ends on an
    interrupted `turn/end` — possibly the reader's in-memory synthesis for a
    turn another process is still writing — in which case the whole trailing
    turn (its `turn/start` on) is withheld for a later poll to prove out."""
    if not entries:
        return entries
    last = event_of(entries[-1])
    if last is None or last.get("type") != "turn/end":
        return entries
    if reason_kind(last) != "interrupted":
        return entries
    for index in range(len(entries) - 1, -1, -1):
        event = event_of(entries[index])
        if event is not None and event.get("type") == "turn/start":
            return entries[:index]
    return []

session_origin 

session_origin(
    client: DshHistoryClient, session_id: str
) -> str | None
Source code in cli/octomate_cli/streaming/deepseek.py 
280
281
282
283
284
285
286
287
288
289
290
291
def session_origin(client: DshHistoryClient, session_id: str) -> str | None:
    value = client.rpc("session/list", {"_request": {}})
    if not isinstance(value, dict):
        return None
    items = value.get("items")
    if not isinstance(items, list):
        return None
    for item in items:
        if isinstance(item, dict) and item.get("sessionId") == session_id:
            origin = item.get("origin")
            return origin if isinstance(origin, str) else None
    return None

stream_session async 

stream_session(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    dsh_url: str,
) -> bool

One connection's life. True when the session is done (the server said finalize, or it went idle); False to reconnect and resume.

Source code in cli/octomate_cli/streaming/deepseek.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
async def stream_session(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    dsh_url: str,
) -> bool:
    """One connection's life. True when the session is done (the server said
    finalize, or it went idle); False to reconnect and resume."""
    history = await asyncio.to_thread(DshHistoryClient, dsh_url)
    async with connect(
        url, additional_headers={"Authorization": f"Bearer {token}"}
    ) as websocket:
        await websocket.send(
            StreamHello(
                protocol=STREAM_PROTOCOL,
                session_id=session_id,
                transcript_path=str(transcript_path),
                cwd=cwd,
                client_version=CLIENT_VERSION,
            ).model_dump_json()
        )
        welcome = server_message_adapter.validate_json(await websocket.recv())
        if not isinstance(welcome, StreamWelcome):
            raise RuntimeError(f"expected a welcome message, got {welcome.type!r}")
        cursor = welcome.offsets.get(SESSION_FILE, 0)

        stop = asyncio.Event()
        finalizing = False

        async def receive_server() -> None:
            nonlocal finalizing
            while True:
                try:
                    raw = await websocket.recv()
                except ConnectionClosed:
                    stop.set()
                    return
                if isinstance(
                    server_message_adapter.validate_json(raw), StreamFinalize
                ):
                    finalizing = True
                    stop.set()
                    return

        async def pump() -> bool:
            nonlocal cursor
            entries = await asyncio.to_thread(new_entries, history, session_id, cursor)
            sent = False
            for entry in shippable(entries):
                seq = seq_of(entry)
                if seq is None or seq < cursor:
                    continue
                if seq > cursor:
                    # The gateway skipped seqs it should serve densely; a
                    # shipped gap would close the stream at 4000, so resync
                    # by reconnecting instead.
                    raise RuntimeError(
                        f"seq gap from dsh: expected {cursor}, got {seq}"
                    )
                await websocket.send(
                    StreamLine(
                        agent_id=None,
                        start=seq,
                        end=seq + 1,
                        line=json_object_adapter.dump_json(entry).decode(),
                    ).model_dump_json()
                )
                cursor = seq + 1
                sent = True
            return sent

        receiver = asyncio.create_task(receive_server())
        last_active = monotonic()
        try:
            while True:
                if await pump():
                    last_active = monotonic()
                if stop.is_set():
                    break
                if monotonic() - last_active > IDLE_TIMEOUT:
                    finalizing = True
                    break
                with contextlib.suppress(TimeoutError):
                    async with asyncio.timeout(POLL_INTERVAL):
                        await stop.wait()
            if not finalizing:
                return False  # the socket dropped: reconnect and resume
            receiver.cancel()
            await pump()  # final drain past the finalize
            await websocket.send(StreamEof().model_dump_json())
            with contextlib.suppress(ConnectionClosed, TimeoutError):
                async with asyncio.timeout(10):
                    await websocket.recv()
            return True
        finally:
            receiver.cancel()
            with contextlib.suppress(ConnectionClosed, asyncio.CancelledError):
                await receiver

run_tail async 

run_tail(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    dsh_url: str,
) -> None

The reconnect loop around stream_session, mirroring the file tails': resume after drops, back off while the server is unreachable, stop once the session goes quiet — the next prompt's launcher starts a fresh tail.

Source code in cli/octomate_cli/streaming/deepseek.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
async def run_tail(
    url: str,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    token: str,
    dsh_url: str,
) -> None:
    """The reconnect loop around `stream_session`, mirroring the file tails':
    resume after drops, back off while the server is unreachable, stop once
    the session goes quiet — the next prompt's launcher starts a fresh tail."""
    attempt = 0
    while True:
        try:
            if await stream_session(
                url, session_id, transcript_path, cwd, token, dsh_url
            ):
                return
            attempt = 0
        except ConnectionClosed as closed:
            code = closed.rcvd.code if closed.rcvd is not None else None
            if code == REFUSED:
                reason = closed.rcvd.reason if closed.rcvd is not None else ""
                print(f"octomate: stream refused: {reason}", file=sys.stderr)
                return
            attempt += 1
        except InvalidStatus as denied:
            if denied.response.status_code in {401, 403}:
                print(
                    "octomate: stream denied — a valid API token with hooks scope "
                    "is required. Run `octomate configure --token <api-token>`.",
                    file=sys.stderr,
                )
                return
            attempt += 1
        except DshCompatibilityError as error:
            print(f"octomate: {error}", file=sys.stderr)
            return
        except (OSError, InvalidHandshake, TimeoutError, RuntimeError):
            attempt += 1
        await asyncio.sleep(min(BACKOFF_CAP, float(2**attempt)))

main 

main(
    *,
    session_id: str,
    transcript_path: Path,
    url: str,
    cwd: str,
    dsh_url: str,
) -> None
Source code in cli/octomate_cli/streaming/deepseek.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
def main(
    *,
    session_id: str,
    transcript_path: Path,
    url: str,
    cwd: str,
    dsh_url: str,
) -> None:
    token = cli_settings().token
    if not token:
        print(
            f"octomate: no credential — {CLISettings.env('token')} is unset and the "
            "client config holds none, so this session is not being streamed. "
            "Run `octomate configure --token <api-token>`.",
            file=sys.stderr,
        )
        raise SystemExit(1)
    # A dsh subagent child is its parent's story; the server never sees the
    # session header, so the classification happens here, against the same
    # gateway the events come from.
    try:
        if session_origin(DshHistoryClient(dsh_url), session_id) == "subagent":
            return
    except DshCompatibilityError as error:
        print(f"octomate: {error}", file=sys.stderr)
        raise SystemExit(1) from None
    lock_path = tail_path("deepseek", session_id).with_suffix(".lock")
    with lock_path.open("w") as lock:
        deadline = monotonic() + LOCK_GRACE
        while True:
            try:
                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                if monotonic() >= deadline:
                    return
                time.sleep(LOCK_POLL)
        asyncio.run(run_tail(url, session_id, transcript_path, cwd, token, dsh_url))