Skip to content

Native session hooks and tailers

octomate.tentacles.hooks

FastAPI dependencies that authenticate a runtime's hook request and resolve its sender.

hook_guard 

hook_guard(
    bearers: KnownBearers,
) -> Callable[[str | None], Awaitable[str]]

Authenticate a hook request against API tokens with the hooks scope.

The dependency yields the owner's username for ledger attribution.

Source code in octomate/tentacles/hooks.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def hook_guard(
    bearers: KnownBearers,
) -> Callable[[str | None], Awaitable[str]]:
    """Authenticate a hook request against API tokens with the hooks scope.

    The dependency yields the owner's username for ledger attribution.
    """

    async def verify(
        authorization: Annotated[str | None, Header()] = None,
    ) -> str:
        username = (
            await bearers.owner(authorization.removeprefix("Bearer "), scope="hooks")
            if authorization is not None and authorization.startswith("Bearer ")
            else None
        )
        if username is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="invalid hook credentials",
            )
        return username

    return verify

hook_sender 

hook_sender(
    users: UserManager,
    runtime: str,
    verify: Callable[[str | None], Awaitable[str]],
) -> Callable[[str], Awaitable[UserProfile]]

The guard's second half, as a FastAPI dependency: the verified principal resolved to their own profile on runtime's pseudo-channel — who every ledger row this request writes is attributed to.

Composes over verify rather than re-reading the header: FastAPI caches a dependency's value per request, so a router-level guard and this one share a single bearer check.

Source code in octomate/tentacles/hooks.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def hook_sender(
    users: UserManager,
    runtime: str,
    verify: Callable[[str | None], Awaitable[str]],
) -> Callable[[str], Awaitable[UserProfile]]:
    """The guard's second half, as a FastAPI dependency: the verified principal
    resolved to their own profile on `runtime`'s pseudo-channel — who every
    ledger row this request writes is attributed to.

    Composes over `verify` rather than re-reading the header: FastAPI caches a
    dependency's value per request, so a router-level guard and this one share a
    single bearer check."""

    async def sender(username: str = Depends(verify)) -> UserProfile:
        profile = await users.native_profile(runtime, username)
        if profile is None:
            # The guard already proved the username names a registered user, so
            # a miss is a wiring bug worth an error, never a row written to nobody.
            raise RuntimeError(
                f"verified bearer names {username!r}, but the registry holds no "
                "such user"
            )
        return profile

    return sender

octomate.tentacles.claude.ingest

Hook-driven ingest of a native Claude Code session.

Hooks establish session and run identities and signal transcript boundaries. Only the transcript tail writes messages.

logger module-attribute 

logger = logging.getLogger(__name__)

ClaudeHookIngest

Establish native session/run identities and relay transcript boundaries.

Hooks may arrive before or after their transcript. They ensure empty runs under the same native turn keys, but never write messages or replace a run's contents. The tail alone creates model messages and their ledger bindings. Its authenticated stream supplies the sender. A hook's transcript path is client context, never a server-local file to open.

Source code in octomate/tentacles/claude/ingest.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 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
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
class ClaudeHookIngest:
    """Establish native session/run identities and relay transcript boundaries.

    Hooks may arrive before or after their transcript. They ensure empty runs
    under the same native turn keys, but never write messages or replace a run's
    contents. The tail alone creates model messages and their ledger bindings.
    Its authenticated stream supplies the sender. A hook's transcript path is
    client context, never a server-local file to open.
    """

    conversations: ConversationManager
    projects: ProjectManager
    threads: ThreadManager
    tailer: ClaudeTranscriptTailer
    locks: SessionLocks

    def __init__(
        self,
        tailer: ClaudeTranscriptTailer,
        locks: SessionLocks | None = None,
        *,
        conversations: ConversationManager,
        projects: ProjectManager,
        threads: ThreadManager,
    ) -> None:
        self.conversations = conversations
        self.projects = projects
        self.threads = threads
        self.tailer = tailer
        # Serialize a session's events so the existence check and the write can't race
        # (Claude fires the next event without waiting for our commit). Shared with the
        # tailer so its run/ledger commits serialize against these writes too; the
        # registry reclaims a session's lock on its own once no one holds it.
        self.locks = locks if locks is not None else SessionLocks()

    async def handle(self, event: ClaudeHookInput) -> None:
        """Handle an authenticated hook; message attribution belongs to the stream."""
        if (
            event.agent_id is None
            and event.prompt_id
            and await self.conversations.driven_run(
                CLAUDE_NATIVE_ID, event.session_id, event.prompt_id
            )
            is not None
        ):
            if event.hook_event_name == "Stop":
                await self.tailer.stop_turn(event.session_id, event.prompt_id)
            return
        match event.hook_event_name:
            case "SubagentStart":
                await self.on_subagent_start(event)
            case "SubagentStop":
                await self.on_subagent_stop(event)
            case _ if event.agent_id is not None:
                # Any other event carrying agent_id fired *inside* a subagent (its
                # own Stop, a PreToolUse, …) and is never a parent-turn event.
                # A child's Stop must never drain the parent turn.
                return
            case "UserPromptSubmit":
                await self.on_user_prompt_submit(event)
            case "Stop":
                await self.on_stop(event)
            case "SessionEnd":
                await self.on_session_end(event)

    @claude_logfire.instrument(
        "claude.hook UserPromptSubmit [{event.session_id}]", extract_args=["event"]
    )
    async def on_user_prompt_submit(self, event: ClaudeHookInput) -> None:
        async with self.locks.hold(event.session_id):
            await self.start_session(event)
            await self.ensure_run(event)

    @claude_logfire.instrument(
        "claude.hook Stop [{event.session_id}]", extract_args=["event"]
    )
    async def on_stop(self, event: ClaudeHookInput) -> None:
        async with self.locks.hold(event.session_id):
            await self.start_session(event)
            await self.ensure_run(event)
        # The turn is over and its transcript flushed: let the tail commit it now
        # rather than at the next prompt — a remote one drains out and exits. Outside
        # the lock, because a local close takes the same session lock to commit.
        await self.tailer.stop_turn(event.session_id, event.prompt_id)

    @claude_logfire.instrument(
        "claude.hook SessionEnd [{event.session_id}]", extract_args=["event"]
    )
    async def on_session_end(self, event: ClaudeHookInput) -> None:
        logger.info("session %s: ended", event.session_id)
        # The thread first: for a session whose earlier hooks never arrived this is
        # the last event carrying a cwd, and a thread's project is frozen at creation
        # — a backfill tail attaching later would create it unfiled. Already-created
        # sessions resolve to their existing row.
        await self.session_thread(event)
        # Finalize outside any lock: it awaits the stream's drain commits, which take
        # the session lock — holding it here would deadlock.
        await self.tailer.finalize(event.session_id)

    @claude_logfire.instrument(
        "claude.hook SubagentStart [{event.session_id}/{event.agent_id}]",
        extract_args=["event"],
    )
    async def on_subagent_start(self, event: ClaudeHookInput) -> None:
        if not event.agent_id:
            return
        async with self.locks.hold(event.session_id):
            await self.start_session(event)
            if event.prompt_id:
                await self.ensure_subagent_run(event)
        logger.info("session %s: subagent %s started", event.session_id, event.agent_id)

    @claude_logfire.instrument(
        "claude.hook SubagentStop [{event.session_id}/{event.agent_id}]",
        extract_args=["event"],
    )
    async def on_subagent_stop(self, event: ClaudeHookInput) -> None:
        if not event.agent_id:
            return
        # The subagent has written its last line by the time this synchronous hook
        # fires — and a Claude child's own file never closes its last turn — so wait
        # for the streamed answer and commit the open turn now. A session no stream
        # covers has nothing to settle.
        await self.tailer.finish_subagent(
            event.session_id,
            event.agent_id,
            final_answer=event.last_assistant_message,
        )
        logger.info("session %s: subagent %s stopped", event.session_id, event.agent_id)

    async def ensure_subagent_run(self, event: ClaudeHookInput) -> None:
        """Ensure the child's empty run without copying its hook prompt."""
        # Both guarded by the caller.
        assert event.agent_id
        assert event.prompt_id
        thread = await self.session_thread(event)
        parent = await self.conversations.ensure(
            thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID
        )
        child = await self.conversations.ensure(
            thread.id,
            agent_tentacle_id=CLAUDE_NATIVE_ID,
            subagent_id=event.agent_id,
            parent_conversation_id=parent.id,
        )
        await self.conversations.record_external_run(
            child,
            run_id=f"{event.agent_id}:{event.prompt_id}",
            name=CLAUDE_NATIVE_ID,
            cwd=Path(event.cwd) if event.cwd else None,
            native_session_id=event.agent_id,
            parent_run_id=event.prompt_id,
        )

    async def start_session(self, event: ClaudeHookInput) -> None:
        """Ensure the session's skeleton (thread + conversation), under the session
        lock so the write can't race the hooks' first-sighting create. The transcript
        itself arrives only through the stream; nothing here follows the path a hook
        claims."""
        thread = await self.session_thread(event)
        await self.conversations.ensure(thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID)

    async def session_thread(self, event: ClaudeHookInput) -> Thread:
        """This session's thread, filed under the project already holding the directory
        it is running in — and under none when no project holds it.

        The session's own directory is what says which project this is, not the
        transcript path, which lives in Claude's tree whatever the code does. Nothing is
        registered from here: Claude's own `projects` are a context store keyed by
        directory, not a claim that a tree is worked in, so a Claude session is not
        evidence of a project the way a Codex workspace is. The directory is recorded on
        the run regardless, which is what a later promotion would read.

        The project lands only when the thread is created, so resolving it on every
        later hook costs a path comparison and rewrites nothing.
        """
        # Only a cwd the hook actually carried: `Path("")` is the process's own
        # directory, which would attribute every session to whatever project
        # Octomate itself was started in.
        holder = self.projects.resolve(Path(event.cwd)) if event.cwd else None
        project = self.projects.get(holder) if holder is not None else None
        return await self.threads.ensure(
            ThreadKey(
                channel_tentacle_id=CLAUDE_NATIVE_ID,
                chat_type="thread",
                chat_id=event.session_id,
                channel_thread_id=None,
            ),
            project=project,
        )

    async def ensure_run(self, event: ClaudeHookInput) -> None:
        """Reserve the turn's foreign-key target; repeated or late hooks change nothing."""
        if not event.prompt_id:
            return
        thread = await self.session_thread(event)
        conversation = await self.conversations.ensure(
            thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID
        )
        await self.conversations.record_external_run(
            conversation,
            run_id=event.prompt_id,
            name=CLAUDE_NATIVE_ID,
            cwd=Path(event.cwd) if event.cwd else None,
            native_session_id=event.session_id,
        )

conversations instance-attribute 

conversations: ConversationManager = conversations

projects instance-attribute 

projects: ProjectManager = projects

threads instance-attribute 

threads: ThreadManager = threads

tailer instance-attribute 

tailer: ClaudeTranscriptTailer = tailer

locks instance-attribute 

locks: SessionLocks = (
    locks if locks is not None else SessionLocks()
)

handle async 

handle(event: ClaudeHookInput) -> None

Handle an authenticated hook; message attribution belongs to the stream.

Source code in octomate/tentacles/claude/ingest.py 
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
async def handle(self, event: ClaudeHookInput) -> None:
    """Handle an authenticated hook; message attribution belongs to the stream."""
    if (
        event.agent_id is None
        and event.prompt_id
        and await self.conversations.driven_run(
            CLAUDE_NATIVE_ID, event.session_id, event.prompt_id
        )
        is not None
    ):
        if event.hook_event_name == "Stop":
            await self.tailer.stop_turn(event.session_id, event.prompt_id)
        return
    match event.hook_event_name:
        case "SubagentStart":
            await self.on_subagent_start(event)
        case "SubagentStop":
            await self.on_subagent_stop(event)
        case _ if event.agent_id is not None:
            # Any other event carrying agent_id fired *inside* a subagent (its
            # own Stop, a PreToolUse, …) and is never a parent-turn event.
            # A child's Stop must never drain the parent turn.
            return
        case "UserPromptSubmit":
            await self.on_user_prompt_submit(event)
        case "Stop":
            await self.on_stop(event)
        case "SessionEnd":
            await self.on_session_end(event)

on_user_prompt_submit async 

on_user_prompt_submit(event: ClaudeHookInput) -> None
Source code in octomate/tentacles/claude/ingest.py 
 98
 99
100
101
102
103
104
@claude_logfire.instrument(
    "claude.hook UserPromptSubmit [{event.session_id}]", extract_args=["event"]
)
async def on_user_prompt_submit(self, event: ClaudeHookInput) -> None:
    async with self.locks.hold(event.session_id):
        await self.start_session(event)
        await self.ensure_run(event)

on_stop async 

on_stop(event: ClaudeHookInput) -> None
Source code in octomate/tentacles/claude/ingest.py 
106
107
108
109
110
111
112
113
114
115
116
@claude_logfire.instrument(
    "claude.hook Stop [{event.session_id}]", extract_args=["event"]
)
async def on_stop(self, event: ClaudeHookInput) -> None:
    async with self.locks.hold(event.session_id):
        await self.start_session(event)
        await self.ensure_run(event)
    # The turn is over and its transcript flushed: let the tail commit it now
    # rather than at the next prompt — a remote one drains out and exits. Outside
    # the lock, because a local close takes the same session lock to commit.
    await self.tailer.stop_turn(event.session_id, event.prompt_id)

on_session_end async 

on_session_end(event: ClaudeHookInput) -> None
Source code in octomate/tentacles/claude/ingest.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
@claude_logfire.instrument(
    "claude.hook SessionEnd [{event.session_id}]", extract_args=["event"]
)
async def on_session_end(self, event: ClaudeHookInput) -> None:
    logger.info("session %s: ended", event.session_id)
    # The thread first: for a session whose earlier hooks never arrived this is
    # the last event carrying a cwd, and a thread's project is frozen at creation
    # — a backfill tail attaching later would create it unfiled. Already-created
    # sessions resolve to their existing row.
    await self.session_thread(event)
    # Finalize outside any lock: it awaits the stream's drain commits, which take
    # the session lock — holding it here would deadlock.
    await self.tailer.finalize(event.session_id)

on_subagent_start async 

on_subagent_start(event: ClaudeHookInput) -> None
Source code in octomate/tentacles/claude/ingest.py 
132
133
134
135
136
137
138
139
140
141
142
143
@claude_logfire.instrument(
    "claude.hook SubagentStart [{event.session_id}/{event.agent_id}]",
    extract_args=["event"],
)
async def on_subagent_start(self, event: ClaudeHookInput) -> None:
    if not event.agent_id:
        return
    async with self.locks.hold(event.session_id):
        await self.start_session(event)
        if event.prompt_id:
            await self.ensure_subagent_run(event)
    logger.info("session %s: subagent %s started", event.session_id, event.agent_id)

on_subagent_stop async 

on_subagent_stop(event: ClaudeHookInput) -> None
Source code in octomate/tentacles/claude/ingest.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
@claude_logfire.instrument(
    "claude.hook SubagentStop [{event.session_id}/{event.agent_id}]",
    extract_args=["event"],
)
async def on_subagent_stop(self, event: ClaudeHookInput) -> None:
    if not event.agent_id:
        return
    # The subagent has written its last line by the time this synchronous hook
    # fires — and a Claude child's own file never closes its last turn — so wait
    # for the streamed answer and commit the open turn now. A session no stream
    # covers has nothing to settle.
    await self.tailer.finish_subagent(
        event.session_id,
        event.agent_id,
        final_answer=event.last_assistant_message,
    )
    logger.info("session %s: subagent %s stopped", event.session_id, event.agent_id)

ensure_subagent_run async 

ensure_subagent_run(event: ClaudeHookInput) -> None

Ensure the child's empty run without copying its hook prompt.

Source code in octomate/tentacles/claude/ingest.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
async def ensure_subagent_run(self, event: ClaudeHookInput) -> None:
    """Ensure the child's empty run without copying its hook prompt."""
    # Both guarded by the caller.
    assert event.agent_id
    assert event.prompt_id
    thread = await self.session_thread(event)
    parent = await self.conversations.ensure(
        thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID
    )
    child = await self.conversations.ensure(
        thread.id,
        agent_tentacle_id=CLAUDE_NATIVE_ID,
        subagent_id=event.agent_id,
        parent_conversation_id=parent.id,
    )
    await self.conversations.record_external_run(
        child,
        run_id=f"{event.agent_id}:{event.prompt_id}",
        name=CLAUDE_NATIVE_ID,
        cwd=Path(event.cwd) if event.cwd else None,
        native_session_id=event.agent_id,
        parent_run_id=event.prompt_id,
    )

start_session async 

start_session(event: ClaudeHookInput) -> None

Ensure the session's skeleton (thread + conversation), under the session lock so the write can't race the hooks' first-sighting create. The transcript itself arrives only through the stream; nothing here follows the path a hook claims.

Source code in octomate/tentacles/claude/ingest.py 
187
188
189
190
191
192
193
async def start_session(self, event: ClaudeHookInput) -> None:
    """Ensure the session's skeleton (thread + conversation), under the session
    lock so the write can't race the hooks' first-sighting create. The transcript
    itself arrives only through the stream; nothing here follows the path a hook
    claims."""
    thread = await self.session_thread(event)
    await self.conversations.ensure(thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID)

session_thread async 

session_thread(event: ClaudeHookInput) -> Thread

This session's thread, filed under the project already holding the directory it is running in — and under none when no project holds it.

The session's own directory is what says which project this is, not the transcript path, which lives in Claude's tree whatever the code does. Nothing is registered from here: Claude's own projects are a context store keyed by directory, not a claim that a tree is worked in, so a Claude session is not evidence of a project the way a Codex workspace is. The directory is recorded on the run regardless, which is what a later promotion would read.

The project lands only when the thread is created, so resolving it on every later hook costs a path comparison and rewrites nothing.

Source code in octomate/tentacles/claude/ingest.py 
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
async def session_thread(self, event: ClaudeHookInput) -> Thread:
    """This session's thread, filed under the project already holding the directory
    it is running in — and under none when no project holds it.

    The session's own directory is what says which project this is, not the
    transcript path, which lives in Claude's tree whatever the code does. Nothing is
    registered from here: Claude's own `projects` are a context store keyed by
    directory, not a claim that a tree is worked in, so a Claude session is not
    evidence of a project the way a Codex workspace is. The directory is recorded on
    the run regardless, which is what a later promotion would read.

    The project lands only when the thread is created, so resolving it on every
    later hook costs a path comparison and rewrites nothing.
    """
    # Only a cwd the hook actually carried: `Path("")` is the process's own
    # directory, which would attribute every session to whatever project
    # Octomate itself was started in.
    holder = self.projects.resolve(Path(event.cwd)) if event.cwd else None
    project = self.projects.get(holder) if holder is not None else None
    return await self.threads.ensure(
        ThreadKey(
            channel_tentacle_id=CLAUDE_NATIVE_ID,
            chat_type="thread",
            chat_id=event.session_id,
            channel_thread_id=None,
        ),
        project=project,
    )

ensure_run async 

ensure_run(event: ClaudeHookInput) -> None

Reserve the turn's foreign-key target; repeated or late hooks change nothing.

Source code in octomate/tentacles/claude/ingest.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
async def ensure_run(self, event: ClaudeHookInput) -> None:
    """Reserve the turn's foreign-key target; repeated or late hooks change nothing."""
    if not event.prompt_id:
        return
    thread = await self.session_thread(event)
    conversation = await self.conversations.ensure(
        thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID
    )
    await self.conversations.record_external_run(
        conversation,
        run_id=event.prompt_id,
        name=CLAUDE_NATIVE_ID,
        cwd=Path(event.cwd) if event.cwd else None,
        native_session_id=event.session_id,
    )

octomate.tentacles.locks

Per-session asyncio locks that clean up after themselves.

SessionLocks

Per-session asyncio.Locks. Not a lock of its own — a thin registry over the stdlib lock, adding the keying the stdlib has no primitive for, plus self-cleanup: each entry is a weak reference, so once no task holds or waits on a key's lock it is reclaimed and the key leaves nothing behind.

async with locks.hold(key) releases on every exit — return, exception, or cancellation — so an acquire cancelled by its timeout frees the lock automatically, no manual dismissal or force-release.

Source code in octomate/tentacles/locks.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class SessionLocks:
    """Per-session `asyncio.Lock`s. Not a lock of its own — a thin registry over the
    stdlib lock, adding the keying the stdlib has no primitive for, plus self-cleanup:
    each entry is a weak reference, so once no task holds or waits on a key's lock it is
    reclaimed and the key leaves nothing behind.

    `async with locks.hold(key)` releases on every exit — return, exception, or
    cancellation — so an acquire cancelled by its timeout frees the lock automatically,
    no manual dismissal or force-release.
    """

    def __init__(self) -> None:
        self.by_session: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()

    @asynccontextmanager
    async def hold(
        self,
        key: str,
        *,
        acquire_timeout: float | None = None,
    ) -> AsyncGenerator[None]:
        """Hold the key's lock for the block. `acquire_timeout` bounds the wait for the
        lock and nothing else: a waiter stuck behind a wedged holder raises
        `TimeoutError` rather than blocking forever, while the block it guards runs
        unbounded once entered. A caller wanting to bound the work too wraps the whole
        `async with` in its own cancel scope; a never-acquired lock is never released."""
        # The local `lock` is a strong reference held for the whole block, so the weak
        # entry stays alive while anyone holds or waits on it — concurrent users share
        # the one live lock — and is reclaimed once the last user leaves.
        lock = self.by_session.get(key)
        if lock is None:
            lock = asyncio.Lock()
            self.by_session[key] = lock
        await asyncio.wait_for(lock.acquire(), acquire_timeout)
        try:
            yield
        finally:
            lock.release()

by_session instance-attribute 

by_session: WeakValueDictionary[str, Lock] = (
    WeakValueDictionary()
)

hold async 

hold(
    key: str, *, acquire_timeout: float | None = None
) -> AsyncGenerator[None]

Hold the key's lock for the block. acquire_timeout bounds the wait for the lock and nothing else: a waiter stuck behind a wedged holder raises TimeoutError rather than blocking forever, while the block it guards runs unbounded once entered. A caller wanting to bound the work too wraps the whole async with in its own cancel scope; a never-acquired lock is never released.

Source code in octomate/tentacles/locks.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@asynccontextmanager
async def hold(
    self,
    key: str,
    *,
    acquire_timeout: float | None = None,
) -> AsyncGenerator[None]:
    """Hold the key's lock for the block. `acquire_timeout` bounds the wait for the
    lock and nothing else: a waiter stuck behind a wedged holder raises
    `TimeoutError` rather than blocking forever, while the block it guards runs
    unbounded once entered. A caller wanting to bound the work too wraps the whole
    `async with` in its own cancel scope; a never-acquired lock is never released."""
    # The local `lock` is a strong reference held for the whole block, so the weak
    # entry stays alive while anyone holds or waits on it — concurrent users share
    # the one live lock — and is reclaimed once the last user leaves.
    lock = self.by_session.get(key)
    if lock is None:
        lock = asyncio.Lock()
        self.by_session[key] = lock
    await asyncio.wait_for(lock.acquire(), acquire_timeout)
    try:
        yield
    finally:
        lock.release()

octomate.tentacles.tailer

The shared lifecycle of a native turn published from a transcript.

NativeTurn dataclass

One transcript turn's live events and prompt bindings.

Tailers supply native boundaries, source message identities, and their own accumulator's message list. This owner rejects committed or driven turns, records prompts in that list, publishes the fold, and settles only after the caller commits the run. Subagent transcripts do not publish on this surface.

Source code in octomate/tentacles/tailer.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 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
@dataclass
class NativeTurn:
    """One transcript turn's live events and prompt bindings.

    Tailers supply native boundaries, source message identities, and their own
    accumulator's message list. This owner rejects committed or driven turns,
    records prompts in that list, publishes the fold, and settles only after the
    caller commits the run. Subagent transcripts do not publish on this surface.
    """

    threads: ThreadManager
    locks: SessionLocks
    address: ChannelAddress
    sender: UserProfile
    impulses: ObjectSendStream[Impulse]
    messages: list[ModelMessage]
    recorded: set[str]
    # Keep the actual requests: list positions can change while folding.
    prompts: dict[UUID7, ModelRequest] = field(default_factory=dict)

    @classmethod
    async def open(
        cls,
        *,
        run_id: str,
        recorded: set[str],
        conversations: ConversationManager,
        threads: ThreadManager,
        locks: SessionLocks,
        address: ChannelAddress,
        sender: UserProfile,
        impulses: ObjectSendStream[Impulse],
        messages: list[ModelMessage],
    ) -> NativeTurn | None:
        """Publish the opening bracket unless this turn already has an owner."""
        if run_id in recorded:
            return None
        if (
            await conversations.driven_run(
                address.channel_tentacle_id, address.chat_id, run_id
            )
            is not None
        ):
            recorded.add(run_id)
            return None
        turn = cls(threads, locks, address, sender, impulses, messages, recorded)
        await impulses.send(
            Impulse(RunStartedEvent(address=address, sender=sender), address)
        )
        return turn

    async def prompt(
        self,
        *,
        message_id: str,
        text: str,
        timestamp: datetime,
        metadata: JsonObject | None = None,
    ) -> None:
        """Record a source prompt once and deliver its request at this point.

        The ledger row grants access before the run is complete. Its id is also
        the enqueue id, so the console reconciles the live prompt with that row.
        """
        async with self.locks.hold(self.address.chat_id):
            inbound = await self.threads.record_native_prompt(
                self.address,
                message_id=message_id,
                text=text,
                sender=self.sender,
                timestamp=timestamp,
            )
        request = ModelRequest(
            parts=[UserPromptPart(content=text, timestamp=timestamp)],
            timestamp=timestamp,
            metadata=metadata,
        )
        self.messages.append(request)
        self.prompts[inbound.id] = request
        await self.impulses.send(
            Impulse(
                EnqueuedMessagesEvent(enqueue_id=str(inbound.id), messages=(request,)),
                self.address,
            )
        )

    async def fold(self, events: Iterable[StreamEvents[str]]) -> None:
        """Consume and publish the accumulator's events on the native thread."""
        for event in events:
            if (wire := to_wire(event)) is not None:
                await self.impulses.send(Impulse(wire, self.address))

    async def close(
        self, run: ExternalAgentRun | None, *, run_id: str, answer: str
    ) -> None:
        """Bind the committed history before publishing the closing bracket.

        The caller holds the session lock through its commit and this binding.
        `record_external_run` preserves input order; resolve the request objects
        against that committed list here instead of tracking indices in tailers.
        A failed commit or binding propagates without publishing `run_ended`.
        """
        if run is None:
            self.recorded.add(run_id)
            return
        sources = {id(request): row_id for row_id, request in self.prompts.items()}
        prompts = {
            sources[id(message)]: stored.id
            for message, stored in zip(self.messages, run.messages, strict=True)
            if id(message) in sources
        }
        if len(prompts) != len(self.prompts):
            raise ValueError("A native prompt disappeared from the committed history")
        reply_id = await self.threads.bind_native_run(
            self.address, run, prompts=prompts, answer=answer, sender=self.sender
        )
        self.recorded.add(run_id)
        await self.impulses.send(
            Impulse(RunEndedEvent(thread_message_id=reply_id), self.address)
        )

threads instance-attribute 

threads: ThreadManager

locks instance-attribute 

locks: SessionLocks

address instance-attribute 

address: ChannelAddress

sender instance-attribute 

sender: UserProfile

impulses instance-attribute 

impulses: ObjectSendStream[Impulse]

messages instance-attribute 

messages: list[ModelMessage]

recorded instance-attribute 

recorded: set[str]

prompts class-attribute instance-attribute 

prompts: dict[UUID7, ModelRequest] = field(
    default_factory=dict
)

open async classmethod 

open(
    *,
    run_id: str,
    recorded: set[str],
    conversations: ConversationManager,
    threads: ThreadManager,
    locks: SessionLocks,
    address: ChannelAddress,
    sender: UserProfile,
    impulses: ObjectSendStream[Impulse],
    messages: list[ModelMessage],
) -> NativeTurn | None

Publish the opening bracket unless this turn already has an owner.

Source code in octomate/tentacles/tailer.py 
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
@classmethod
async def open(
    cls,
    *,
    run_id: str,
    recorded: set[str],
    conversations: ConversationManager,
    threads: ThreadManager,
    locks: SessionLocks,
    address: ChannelAddress,
    sender: UserProfile,
    impulses: ObjectSendStream[Impulse],
    messages: list[ModelMessage],
) -> NativeTurn | None:
    """Publish the opening bracket unless this turn already has an owner."""
    if run_id in recorded:
        return None
    if (
        await conversations.driven_run(
            address.channel_tentacle_id, address.chat_id, run_id
        )
        is not None
    ):
        recorded.add(run_id)
        return None
    turn = cls(threads, locks, address, sender, impulses, messages, recorded)
    await impulses.send(
        Impulse(RunStartedEvent(address=address, sender=sender), address)
    )
    return turn

prompt async 

prompt(
    *,
    message_id: str,
    text: str,
    timestamp: datetime,
    metadata: JsonObject | None = None,
) -> None

Record a source prompt once and deliver its request at this point.

The ledger row grants access before the run is complete. Its id is also the enqueue id, so the console reconciles the live prompt with that row.

Source code in octomate/tentacles/tailer.py 
 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
async def prompt(
    self,
    *,
    message_id: str,
    text: str,
    timestamp: datetime,
    metadata: JsonObject | None = None,
) -> None:
    """Record a source prompt once and deliver its request at this point.

    The ledger row grants access before the run is complete. Its id is also
    the enqueue id, so the console reconciles the live prompt with that row.
    """
    async with self.locks.hold(self.address.chat_id):
        inbound = await self.threads.record_native_prompt(
            self.address,
            message_id=message_id,
            text=text,
            sender=self.sender,
            timestamp=timestamp,
        )
    request = ModelRequest(
        parts=[UserPromptPart(content=text, timestamp=timestamp)],
        timestamp=timestamp,
        metadata=metadata,
    )
    self.messages.append(request)
    self.prompts[inbound.id] = request
    await self.impulses.send(
        Impulse(
            EnqueuedMessagesEvent(enqueue_id=str(inbound.id), messages=(request,)),
            self.address,
        )
    )

fold async 

fold(events: Iterable[StreamEvents[str]]) -> None

Consume and publish the accumulator's events on the native thread.

Source code in octomate/tentacles/tailer.py 
120
121
122
123
124
async def fold(self, events: Iterable[StreamEvents[str]]) -> None:
    """Consume and publish the accumulator's events on the native thread."""
    for event in events:
        if (wire := to_wire(event)) is not None:
            await self.impulses.send(Impulse(wire, self.address))

close async 

close(
    run: ExternalAgentRun | None,
    *,
    run_id: str,
    answer: str,
) -> None

Bind the committed history before publishing the closing bracket.

The caller holds the session lock through its commit and this binding. record_external_run preserves input order; resolve the request objects against that committed list here instead of tracking indices in tailers. A failed commit or binding propagates without publishing run_ended.

Source code in octomate/tentacles/tailer.py 
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
async def close(
    self, run: ExternalAgentRun | None, *, run_id: str, answer: str
) -> None:
    """Bind the committed history before publishing the closing bracket.

    The caller holds the session lock through its commit and this binding.
    `record_external_run` preserves input order; resolve the request objects
    against that committed list here instead of tracking indices in tailers.
    A failed commit or binding propagates without publishing `run_ended`.
    """
    if run is None:
        self.recorded.add(run_id)
        return
    sources = {id(request): row_id for row_id, request in self.prompts.items()}
    prompts = {
        sources[id(message)]: stored.id
        for message, stored in zip(self.messages, run.messages, strict=True)
        if id(message) in sources
    }
    if len(prompts) != len(self.prompts):
        raise ValueError("A native prompt disappeared from the committed history")
    reply_id = await self.threads.bind_native_run(
        self.address, run, prompts=prompts, answer=answer, sender=self.sender
    )
    self.recorded.add(run_id)
    await self.impulses.send(
        Impulse(RunEndedEvent(thread_message_id=reply_id), self.address)
    )

octomate.tentacles.claude.tailer

Assembly of native Claude Code turns from a streamed transcript.

Lines streamed in by octomate claude tail feed ClaudeTranscriptTailer, which commits each completed turn as an ExternalAgentRun.

logger module-attribute 

logger = logging.getLogger(__name__)

COMMIT_LOCK_TIMEOUT module-attribute 

COMMIT_LOCK_TIMEOUT = 30.0

SUBAGENT_SETTLE_TIMEOUT module-attribute 

SUBAGENT_SETTLE_TIMEOUT = 2.0

SUBAGENT_SETTLE_POLL module-attribute 

SUBAGENT_SETTLE_POLL = 0.2

REMOTE_DRAIN_TIMEOUT module-attribute 

REMOTE_DRAIN_TIMEOUT = 5.0

OpenTurn dataclass

The turn currently being assembled off the live tail — its accumulator fed line by line, its byte range advancing to cover the last transcript line folded in. It commits as an ExternalAgentRun when the next prompt line or its Stop hooks' summary line closes it, when its own Stop hook reaches stop_turn, or at finalize.

Source code in octomate/tentacles/claude/tailer.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@dataclass
class OpenTurn:
    """The turn currently being assembled off the live tail — its accumulator fed line
    by line, its byte range advancing to cover the last transcript line folded in. It
    commits as an `ExternalAgentRun` when the next prompt line or its `Stop` hooks'
    summary line closes it, when its own `Stop` hook reaches `stop_turn`, or at
    `finalize`."""

    prompt_id: str
    source: str | None  # the transcript `entrypoint` (claude-vscode / cli / …)
    # Where the turn's opening prompt line says it ran — the opening line specifically,
    # because a transcript carries a `cwd` on every line and a turn's tools wander into
    # scratchpads and site-packages, so the last line folded in is not where the run
    # happened. Per turn rather than per session, because a session resumed from another
    # directory carries on in that one.
    cwd: str
    start_offset: int  # byte offset of this turn's opening prompt line
    end_offset: int  # byte offset past the last line folded in
    accumulator: ClaudeRunAccumulator
    last_line_uuid: str | None
    permission_mode: str | None  # Permission mode on the turn's opening prompt.
    native: NativeTurn | None = None

prompt_id instance-attribute 

prompt_id: str

source instance-attribute 

source: str | None

cwd instance-attribute 

cwd: str

start_offset instance-attribute 

start_offset: int

end_offset instance-attribute 

end_offset: int

accumulator instance-attribute 

accumulator: ClaudeRunAccumulator

last_line_uuid instance-attribute 

last_line_uuid: str | None

permission_mode instance-attribute 

permission_mode: str | None

native class-attribute instance-attribute 

native: NativeTurn | None = None

SubagentTail dataclass

One subagent transcript's cursor within a session's tail. A subagent writes its own file (<session>/subagents/agent-<agentId>.jsonl), so it gets its own byte cursor, conversation, and open turn. Its turns frame on promptId change: subagent files carry no promptSource, and a resumed subagent's next turn may open on a tool-result line, not a prompt.

Source code in octomate/tentacles/claude/tailer.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@dataclass
class SubagentTail:
    """One subagent transcript's cursor within a session's tail. A subagent writes its
    own file (`<session>/subagents/agent-<agentId>.jsonl`), so it gets its own byte
    cursor, conversation, and open turn. Its turns frame on `promptId` change: subagent files
    carry no `promptSource`, and a resumed subagent's next turn may open on a
    tool-result line, not a prompt."""

    agent_id: str
    path: Path
    offset: int = 0
    conversation: Conversation | None = None
    recorded: set[str] = field(default_factory=set)  # child run ids already committed
    open_turn: OpenTurn | None = None

agent_id instance-attribute 

agent_id: str

path instance-attribute 

path: Path

offset class-attribute instance-attribute 

offset: int = 0

conversation class-attribute instance-attribute 

conversation: Conversation | None = None

recorded class-attribute instance-attribute 

recorded: set[str] = field(default_factory=set)

open_turn class-attribute instance-attribute 

open_turn: OpenTurn | None = None

TailState dataclass

One streamed session: its transcript lives on the client machine that streams it (octomate claude tail); the stream route feeds feed_remote, which drives the per-line assembly. transcript_path is the client's claim in the client's own namespace, a label never opened here — as are the subagent tails' paths.

Source code in octomate/tentacles/claude/tailer.py 
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
@dataclass
class TailState:
    """One streamed session: its transcript lives on the client machine that streams
    it (`octomate claude tail`); the stream route feeds `feed_remote`, which drives
    the per-line assembly. `transcript_path` is the client's claim in the client's
    own namespace, a label never opened here — as are the subagent tails' paths."""

    session_id: str
    transcript_path: Path
    stop_event: asyncio.Event
    # The verified bearer's own profile — who every ledger row this stream
    # writes is attributed to.
    sender: UserProfile
    address: ChannelAddress  # The native surface resolved by the tentacle.
    impulses: ObjectSendStream[Impulse]  # The publisher open for this connection.
    offset: int = 0
    conversation: Conversation | None = None
    recorded: set[str] = field(default_factory=set)  # prompt_ids already committed
    open_turn: OpenTurn | None = None
    subagents: dict[str, SubagentTail] = field(default_factory=dict)
    # A parent's file may be ahead of its child, including resumed invocations.
    subagent_calls: dict[tuple[str, str], str] = field(default_factory=dict)
    # The prompt id of a turn whose `Stop` hook fired (`stop_turn`): commit it without
    # waiting for the next prompt line — but commit nothing newer, so a prompt queued
    # into the drain window is left open for the next connect to re-stream.
    drain_turn: str | None = None
    # Set once the session's trailing turns are committed (`finish_remote`), so
    # `finalize` can bound its wait for the client's drain.
    drained: asyncio.Event = field(default_factory=asyncio.Event)

    @property
    def subagents_dir(self) -> Path:
        # `<slug>/<session-id>.jsonl` keeps its subagents at `<slug>/<session-id>/subagents`.
        return self.transcript_path.with_suffix("") / "subagents"

session_id instance-attribute 

session_id: str

transcript_path instance-attribute 

transcript_path: Path

stop_event instance-attribute 

stop_event: Event

sender instance-attribute 

sender: UserProfile

address instance-attribute 

address: ChannelAddress

impulses instance-attribute 

impulses: ObjectSendStream[Impulse]

offset class-attribute instance-attribute 

offset: int = 0

conversation class-attribute instance-attribute 

conversation: Conversation | None = None

recorded class-attribute instance-attribute 

recorded: set[str] = field(default_factory=set)

open_turn class-attribute instance-attribute 

open_turn: OpenTurn | None = None

subagents class-attribute instance-attribute 

subagents: dict[str, SubagentTail] = field(
    default_factory=dict
)

subagent_calls class-attribute instance-attribute 

subagent_calls: dict[tuple[str, str], str] = field(
    default_factory=dict
)

drain_turn class-attribute instance-attribute 

drain_turn: str | None = None

drained class-attribute instance-attribute 

drained: Event = field(default_factory=asyncio.Event)

subagents_dir property 

subagents_dir: Path

ClaudeTranscriptTailer

Assembles Claude turns from transcript lines streamed in by octomate claude tail — the only assembler: the server never opens a transcript, even for a session on this machine. Hooks can reserve empty runs, but only the transcript produces model messages and ledger rows.

Each framed line feeds a ClaudeRunAccumulator — the same translation the live tentacle and the whole-file restore use — driving two decoupled sinks:

  • durable — one ExternalAgentRun per completed turn, carrying the byte range it was built from, filling any empty run reserved by a hook. Older hook sketches are replaced for compatibility. Idempotent by prompt_id.
  • live — wire events sent through the connection's nerve publisher at the native thread address, waiting for no reader. Only uncommitted parent turns publish: run_started, the fold, then run_ended after the reply is bound. A reconnect reopens an unfinished turn with another run_started; no reflex is woken.

A turn commits on a proven boundary: the next prompt line, or the drained eof a relayed Stop/SessionEnd asks the client for — the transcript is flushed before Claude fires those hooks. Nothing newer than a stopped turn commits early (drain_turn), and the tail process exits once the stopped turn is durable; the next prompt's launcher spawns a fresh one. The committed runs' offsets are what a reconnecting client is told to resume from, so a connection that drops mid-stream leaves its open turn for the next connect to re-stream.

Source code in octomate/tentacles/claude/tailer.py 
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
271
272
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
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
394
395
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
437
438
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
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
class ClaudeTranscriptTailer:
    """Assembles Claude turns from transcript lines streamed in by `octomate claude
    tail` — the only assembler: the server never opens a transcript, even for a
    session on this machine. Hooks can reserve empty runs, but only the transcript
    produces model messages and ledger rows.

    Each framed line feeds a `ClaudeRunAccumulator` — the same translation the live
    tentacle and the whole-file restore use — driving two decoupled sinks:

    - **durable** — one `ExternalAgentRun` per completed turn, carrying the byte range it
      was built from, filling any empty run reserved by a hook. Older hook sketches
      are replaced for compatibility. Idempotent by `prompt_id`.
    - **live** — wire events sent through the connection's nerve publisher at the
      native thread address, waiting for no reader. Only uncommitted parent turns publish:
      `run_started`, the fold, then `run_ended` after the reply is bound. A reconnect
      reopens an unfinished turn with another `run_started`; no reflex is woken.

    A turn commits on a proven boundary: the next prompt line, or the drained `eof`
    a relayed `Stop`/`SessionEnd` asks the client for — the transcript is flushed
    before Claude fires those hooks. Nothing newer than a stopped turn commits early
    (`drain_turn`), and the tail process exits once the stopped turn is durable; the
    next prompt's launcher spawns a fresh one. The committed runs' offsets are what
    a reconnecting client is told to resume from, so a connection that drops
    mid-stream leaves its open turn for the next connect to re-stream.
    """

    def __init__(
        self,
        conversation_manager: ConversationManager,
        thread_manager: ThreadManager,
        locks: SessionLocks | None = None,
    ) -> None:
        self.conversation_manager = conversation_manager
        self.thread_manager = thread_manager
        # Per-session locks, shared with the hook ingest when wired together, so a turn's
        # run commit here can't race a hook reserving the same run. Its own registry when the tailer runs standalone.
        self.locks = locks if locks is not None else SessionLocks()
        self.sessions: dict[str, TailState] = {}

    async def stop_turn(self, session_id: str, prompt_id: str | None) -> None:
        """The turn just stopped — its `Stop` hook fired: drain it now instead of
        waiting for the next prompt line to close it.

        The stream route relays `finalize` on `stop_event`, the client ships its
        final lines and answers `eof`, and `finish_remote` commits the stopped turn —
        the tail process exits, and the next prompt's launcher spawns a fresh one.
        Scoped by `prompt_id` so nothing newer than the stopped turn ever commits
        early; without one there is nothing to scope, and the turn keeps its old
        boundary. A session no stream covers has nothing to drain.
        """
        if prompt_id is None:
            return
        state = self.sessions.get(session_id)
        if state is None:
            return
        state.drain_turn = prompt_id
        state.stop_event.set()

    async def finalize(self, session_id: str) -> None:
        """End a session (`SessionEnd`): relay the drain to its stream — the route
        watches `stop_event` and sends the client `finalize`; the client drains,
        answers `eof`, and `finish_remote` commits the trailing turns and sets
        `drained`. The wait is bounded: a client that is gone leaves its turns for
        the next connect, and the SessionEnd hook must not hang on it. A session no
        stream covers has nothing to finalize."""
        state = self.sessions.get(session_id)
        if state is None:
            return
        state.stop_event.set()
        try:
            await asyncio.wait_for(state.drained.wait(), REMOTE_DRAIN_TIMEOUT)
        except TimeoutError:
            logger.warning(
                "session %s: remote tail did not drain within %ss; its trailing "
                "turns are left for the next connect to recover",
                session_id,
                REMOTE_DRAIN_TIMEOUT,
            )

    async def shutdown(self) -> None:
        self.sessions.clear()

    async def attach_remote(
        self,
        session_id: str,
        transcript_path: Path,
        sender: UserProfile,
        address: ChannelAddress,
        impulses: ObjectSendStream[Impulse],
    ) -> tuple[TailState, dict[str, int]]:
        """Register a streamed session and answer where each of its files resumes.

        The offsets are recomputed from the committed runs, per file: the session
        transcript under `SESSION_FILE`, each known subagent
        under its agent id — so the client holds no durable cursor and a reconnect is
        just re-asking. A lingering registration (its client died without a close) is
        replaced; the dead route feeds the state object it attached, and its commits
        are idempotent.
        """
        state = TailState(
            session_id=session_id,
            transcript_path=transcript_path,
            stop_event=asyncio.Event(),
            sender=sender,
            address=address,
            impulses=impulses,
        )
        await self.prepare(state)

        conversation = state.conversation
        if conversation is None:
            raise RuntimeError(f"session {session_id} prepared without a conversation")
        state.offset = max(
            (
                run.end_offset or 0
                for run in conversation.runs
                if isinstance(run, ExternalAgentRun)
            ),
            default=0,
        )
        offsets: dict[str, int] = {SESSION_FILE: state.offset}
        for child in await self.conversation_manager.subagents(conversation.id):
            if not child.subagent_id:
                continue
            tail = self.subagent_tail(state, child.subagent_id)
            tail.conversation = child
            tail.recorded = child.native_run_ids
            tail.offset = max(
                (
                    run.end_offset or 0
                    for run in child.runs
                    if isinstance(run, ExternalAgentRun)
                ),
                default=0,
            )
            offsets[child.subagent_id] = tail.offset
        self.sessions[session_id] = state
        logger.info(
            "session %s: remote tail attached, resuming at %s", session_id, offsets
        )
        return state, offsets

    async def feed_remote(
        self, state: TailState, agent_id: str | None, raw: str, start: int, end: int
    ) -> None:
        """Advance a remote session by one framed line — the pumps' per-line body,
        against offsets the client measured in its own file. Takes the state rather
        than a session id so a superseded connection keeps feeding the state it
        attached, never whoever registered after it."""
        if agent_id is None:
            state.offset = end
            line = self.parse_line(raw, state.session_id)
            if line is not None:
                await self.process_line(state, line, start, end)
            return
        tail = self.subagent_tail(state, agent_id)
        if tail.conversation is None:
            await self.prepare_subagent(state, tail)
        tail.offset = end
        line = self.parse_line(raw, state.session_id)
        if line is not None:
            await self.process_subagent_line(state, tail, line, start, end)

    async def finish_remote(self, state: TailState) -> None:
        """A remote session's clean end — the client drained to EOF and said `eof`.

        After the session's own end (`finalize` relayed at `SessionEnd`, or the idle
        drain), commit everything still open. After a relayed `Stop` (`drain_turn`
        set), commit only the stopped turn: a prompt queued into the drain window is
        left open to re-stream whole on the next connect, and subagent turns keep
        their own boundary — `SubagentStop` commits them, and force-closing here
        would truncate a background child still running across turns. Either way,
        release any finalize waiter and reclaim the registry slot."""
        try:
            if state.drain_turn is None:
                await self.close_turn(state)
                await self.close_subagent_turns(state)
            else:
                turn = state.open_turn
                if turn is not None and turn.prompt_id == state.drain_turn:
                    await self.close_turn(state)
        finally:
            state.drained.set()
            self.detach_remote(state)

    def detach_remote(self, state: TailState) -> None:
        """Drop a remote session's registration without committing anything: the
        connection died mid-stream, so its open turns' bytes were never provably
        complete. The next connect resumes from the committed offsets and re-streams
        them."""
        if self.sessions.get(state.session_id) is state:
            del self.sessions[state.session_id]

    async def prepare(self, state: TailState) -> None:
        """Resolve the session's conversation and seed the committed-turn guard from it,
        so a resume never re-commits a run the last run already wrote."""
        state.conversation = await self.ensure_session(state.session_id)
        state.recorded = state.conversation.native_run_ids

    async def ensure_session(self, session_id: str) -> Conversation:
        """Map a native Claude session id to its Octomate home — the session's thread and
        the conversation hanging off it — creating either if this is its first sighting."""
        thread = await self.thread_manager.ensure(
            ThreadKey(CLAUDE_NATIVE_ID, "thread", session_id)
        )
        return await self.conversation_manager.ensure(
            thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID
        )

    @staticmethod
    def parse_line(raw: str | bytes, session_id: str) -> TranscriptLine | None:
        """One typed line, or None for a blank / malformed / unmodeled one — skipped
        but never allowed to wedge ingest; the caller advances its cursor either way."""
        stripped = raw.strip()
        if not stripped:
            return None
        try:
            return transcript_line_adapter.validate_json(stripped)
        except ValidationError:
            logger.debug(
                "skipping unmodeled/malformed transcript line in session %s",
                session_id,
            )
            return None

    async def process_line(
        self, state: TailState, line: TranscriptLine, start: int, end: int
    ) -> None:
        """Route one typed line: a `prompt_source` user line opens a turn (closing the
        previous one); other user (tool-result) and assistant lines fold into the open
        turn, and a `stop_hook_summary` closes it. Inline `is_sidechain` lines are
        skipped — transcripts since 2.1.177 keep subagents in their own files
        (`pump_subagents`), so this guards only against an older transcript's inline
        relics."""
        if isinstance(line, TranscriptUserLine):
            if line.is_sidechain:
                return
            await self.harvest_subagent_call(state, line)
            if line.prompt_source is not None:
                await self.close_turn(state)
                await self.record_permission_mode(state, line.permission_mode)
                state.open_turn = await self.begin_turn(state, line, start, end)
            elif state.open_turn is not None:
                await self.fold(state, line, end)
        elif isinstance(line, TranscriptAssistantLine):
            if line.is_sidechain:
                return
            if state.open_turn is not None:
                await self.fold(state, line, end)
        elif isinstance(line, TranscriptSystemLine):
            # Claude writes it once the turn's `Stop` hooks have run, delivered or
            # not: a server down through the stop never hears the hook.
            if line.subtype == "stop_hook_summary" and not line.is_sidechain:
                await self.close_turn(state)
        elif isinstance(line, TranscriptAiTitleLine):
            await self.record_title(state, line.ai_title)

    async def record_title(self, state: TailState, title: str) -> None:
        """Carry the name Claude gave this session onto the rows that show it.

        The transcript restates the title on every turn and revises it as the work
        turns out to be about something else, so the comparison — not the write —
        is what runs per line. The thread takes it too: a native thread is its
        session, and a name about the work beats the line the thread opened with.
        """
        conversation = state.conversation
        if conversation is None or conversation.name == title.strip():
            return
        await self.conversation_manager.set_name(conversation, title)
        thread = await self.thread_manager.ensure(
            ThreadKey(CLAUDE_NATIVE_ID, "thread", state.session_id)
        )
        await self.thread_manager.rename(thread, title)

    async def harvest_subagent_call(
        self, state: TailState, line: TranscriptUserLine
    ) -> None:
        """Remember which parent tool call spawned which subagent: an `Agent` call's
        tool-result line names the child (`toolUseResult.agentId`) while its content
        block names the call. Stamps child runs' `parent_tool_call_id` at commit."""
        result = line.tool_use_result
        if not isinstance(result, dict):
            return
        agent_id = result.get("agentId")
        if not isinstance(agent_id, str) or not agent_id:
            return
        content = line.message.content
        if not isinstance(content, list):
            return
        tool_use_id = next(
            (
                block.get("tool_use_id")
                for block in content
                if isinstance(block, dict) and block.get("type") == "tool_result"
            ),
            None,
        )
        if not isinstance(tool_use_id, str):
            return
        async with self.locks.hold(state.session_id):
            state.subagent_calls[agent_id, line.prompt_id] = tool_use_id
            await self.conversation_manager.link_parent_run(
                f"{agent_id}:{line.prompt_id}",
                parent_run_id=line.prompt_id,
                parent_tool_call_id=tool_use_id,
            )

    async def fold(
        self,
        state: TailState,
        line: TranscriptAssistantLine | TranscriptUserLine,
        end: int,
    ) -> None:
        """Fold a line into the open turn: consume it (pushing its live events) and
        extend the turn's byte range and provenance to cover it."""
        turn = state.open_turn
        if turn is None:
            raise RuntimeError(f"session {state.session_id}: no open turn to fold into")
        if turn.native is None:
            return
        written = len(turn.accumulator.messages)
        if isinstance(line, TranscriptUserLine) and (text := prompt_text(line.message)):
            await turn.native.prompt(
                message_id=line.uuid,
                text=text,
                timestamp=line.timestamp,
            )
        await turn.native.fold(turn.accumulator.consume(line))
        stamp(turn.accumulator.messages[written:], line.timestamp)
        turn.end_offset = end
        turn.last_line_uuid = line.uuid

    async def record_permission_mode(
        self, state: TailState, permission_mode: str | None
    ) -> None:
        """Keep the conversation's posture at what the session's own transcript says.

        Every prompt line carries the mode that turn ran under, so a ⇧⇥ in the client
        reaches Octomate on the operator's next message. Observed, never set: nothing
        here can change a running client's mode, and the console shows it read-only.

        A mode Claude has that this build's vocabulary does not is skipped rather than
        stored — the column is validated on read, so an unknown one would take the
        conversation out of circulation instead of merely being unrecognized.
        """
        conversation = state.conversation
        if conversation is None or permission_mode is None:
            return
        if permission_mode == conversation.permission_mode:
            return
        if not is_claude_mode(permission_mode):
            logger.debug(
                "session %s reports permission mode %r, which this build does not "
                "model; leaving the conversation at %r",
                state.session_id,
                permission_mode,
                conversation.permission_mode,
            )
            return
        await self.conversation_manager.set_permission_mode(
            conversation, permission_mode
        )

    async def begin_turn(
        self, state: TailState, line: TranscriptUserLine, start: int, end: int
    ) -> OpenTurn:
        """Open the native lifecycle at the transcript's prompt boundary."""
        accumulator = ClaudeRunAccumulator()
        native = await NativeTurn.open(
            run_id=line.prompt_id,
            recorded=state.recorded,
            conversations=self.conversation_manager,
            threads=self.thread_manager,
            locks=self.locks,
            address=state.address,
            sender=state.sender,
            impulses=state.impulses,
            messages=accumulator.messages,
        )
        text = prompt_text(line.message)
        if native is not None:
            await native.prompt(
                message_id=line.prompt_id, text=text, timestamp=line.timestamp
            )
        return OpenTurn(
            prompt_id=line.prompt_id,
            source=line.entrypoint,
            cwd=line.cwd,
            start_offset=start,
            end_offset=end,
            accumulator=accumulator,
            last_line_uuid=line.uuid,
            permission_mode=line.permission_mode,
            native=native,
        )

    async def close_turn(self, state: TailState) -> None:
        """Commit the open turn as an `ExternalAgentRun` with its byte range, then bind
        the turn's human-ledger rows to it. Idempotent by `prompt_id`, so a re-driven
        overlap after a resume is a safe no-op. Held under the session lock so the commit
        cannot race a hook reserving the same turn.

        A commit that cannot be made propagates instead of being skipped past. A
        reconnect resumes from the last committed turn's `end_offset`, which only
        points at the right bytes while the committed turns are the *earliest* ones:
        skipping a turn and streaming on would let a later turn push that mark past
        the gap, stranding the skipped turn where no re-stream could reach it.
        Failing keeps the mark honest — the next connect re-streams the turn from
        there and commits it.
        """
        turn = state.open_turn
        state.open_turn = None
        if turn is None or turn.native is None:
            return
        conversation = state.conversation
        if conversation is None:
            raise RuntimeError(f"session {state.session_id} has no conversation")
        with claude_logfire.span(
            "claude.tailer.commit_turn {prompt_id} [{session_id}]",
            prompt_id=turn.prompt_id,
            session_id=state.session_id,
            start_offset=turn.start_offset,
            end_offset=turn.end_offset,
            messages=len(turn.accumulator.messages),
        ) as span:
            async with self.locks.hold(
                state.session_id, acquire_timeout=COMMIT_LOCK_TIMEOUT
            ):
                run = await self.conversation_manager.record_external_run(
                    conversation,
                    run_id=turn.prompt_id,
                    messages=turn.accumulator.messages,
                    name=CLAUDE_NATIVE_ID,
                    model_name=turn.accumulator.model_name,
                    permission_mode=turn.permission_mode,
                    cwd=Path(turn.cwd) if turn.cwd else None,
                    native_session_id=state.session_id,
                    source=turn.source,
                    start_offset=turn.start_offset,
                    end_offset=turn.end_offset,
                    last_line_uuid=turn.last_line_uuid,
                )
                span.set_attribute("committed", run is not None)
                await turn.native.close(
                    run, run_id=turn.prompt_id, answer=turn.accumulator.result_text
                )
                logger.info(
                    "session %s: turn %s synced — %d messages, bytes %d-%d",
                    state.session_id,
                    turn.prompt_id,
                    len(turn.accumulator.messages),
                    turn.start_offset,
                    turn.end_offset,
                )

    @staticmethod
    def subagent_tail(state: TailState, agent_id: str) -> SubagentTail:
        tail = state.subagents.get(agent_id)
        if tail is None:
            tail = SubagentTail(
                agent_id=agent_id,
                path=state.subagents_dir / f"agent-{agent_id}.jsonl",
            )
            state.subagents[agent_id] = tail
        return tail

    async def finish_subagent(
        self, session_id: str, agent_id: str, *, final_answer: str | None = None
    ) -> None:
        """Drain and commit one subagent on its `SubagentStop` — the child's
        `finalize`, which a Claude child needs: its turns frame on promptId change,
        so nothing in its own file closes the last one. The hook is synchronous but
        the transcript writer is not: the final answer line races it (measured:
        usually loses by ~1-2KB). The child's lines arrive through `feed_remote` on
        their own; when the event names the answer, wait until the stream yields it,
        bounded so a mismatch can never hang the pipe — `tail.offset` says whether
        the writer has gone quiet. A session no stream covers has nothing to
        settle."""
        state = self.sessions.get(session_id)
        if state is None:
            return
        if state.conversation is None:
            await self.prepare(state)
        tail = self.subagent_tail(state, agent_id)
        # The turn this stop is for. If a resume arrives inside the settle window,
        # the stream itself closes this turn on the promptId change and opens the
        # next — which must NOT be committed by this stop, mid-flight.
        target = tail.open_turn
        deadline = monotonic() + SUBAGENT_SETTLE_TIMEOUT
        quiet = 0
        while not self.subagent_settled(tail, final_answer):
            if quiet >= 2:
                break  # writer went silent: the file is as final as it gets
            if monotonic() >= deadline:
                logger.warning(
                    "session %s: subagent %s kept writing past the settle window "
                    "without yielding its announced answer; committing what it "
                    "reached",
                    session_id,
                    agent_id,
                )
                break
            before = tail.offset
            await asyncio.sleep(SUBAGENT_SETTLE_POLL)
            quiet = quiet + 1 if tail.offset == before else 0
        if target is None or tail.open_turn is target:
            await self.close_subagent_turn(state, tail)

    @staticmethod
    def subagent_settled(tail: SubagentTail, final_answer: str | None) -> bool:
        """The fast path: the drained turn already carries the exact answer the
        stop event announced. Divergence (a truncated or differently-joined hook
        payload — no contract promises byte equality) is not an error: the caller's
        byte-quiescence fallback settles those, so strictness here only ever costs
        one extra poll beat, never a wrong commit."""
        if final_answer is None:
            return False  # nothing announced; quiescence decides
        turn = tail.open_turn
        if turn is None:
            return False
        return turn.accumulator.result_text.strip() == final_answer.strip()

    async def prepare_subagent(self, state: TailState, tail: SubagentTail) -> None:
        """Resolve the child's conversation under the session's thread and seed its
        committed-turn guard — the first time this child's lines are fed."""
        parent = state.conversation
        if parent is None:
            raise RuntimeError(f"session {state.session_id} has no conversation")
        tail.conversation = await self.conversation_manager.ensure(
            parent.thread_id,
            agent_tentacle_id=CLAUDE_NATIVE_ID,
            subagent_id=tail.agent_id,
            parent_conversation_id=parent.id,
        )
        tail.recorded = tail.conversation.native_run_ids

    async def process_subagent_line(
        self,
        state: TailState,
        tail: SubagentTail,
        line: TranscriptLine,
        start: int,
        end: int,
    ) -> None:
        """Frame a subagent's turns on `promptId` change — the only marker its file
        carries (`promptSource` is absent, and every line is `is_sidechain`, so
        neither parent rule applies). Each turn's `promptId` is the *parent* turn
        that drove it; a resumed subagent opens its next turn on whatever user line
        arrives first, which is measured to be a tool result, not a prompt."""
        if isinstance(line, TranscriptUserLine):
            turn = tail.open_turn
            if turn is None or line.prompt_id != turn.prompt_id:
                await self.close_subagent_turn(state, tail)
                tail.open_turn = self.begin_subagent_turn(line, start, end)
            else:
                self.fold_subagent(tail, line, end)
        elif isinstance(line, TranscriptAssistantLine):
            if tail.open_turn is not None:
                self.fold_subagent(tail, line, end)

    def begin_subagent_turn(
        self, line: TranscriptUserLine, start: int, end: int
    ) -> OpenTurn:
        accumulator = ClaudeRunAccumulator()
        text = prompt_text(line.message)
        if text:
            accumulator.begin(text)
        # The opener may itself carry tool results (a resumed subagent's turn opens on
        # one); consuming is a no-op for a plain prompt line.
        for _ in accumulator.consume(line):
            pass
        stamp(accumulator.messages, line.timestamp)
        return OpenTurn(
            prompt_id=line.prompt_id,
            source=line.entrypoint,
            cwd=line.cwd,
            start_offset=start,
            end_offset=end,
            accumulator=accumulator,
            last_line_uuid=line.uuid,
            permission_mode=line.permission_mode,
        )

    def fold_subagent(
        self,
        tail: SubagentTail,
        line: TranscriptAssistantLine | TranscriptUserLine,
        end: int,
    ) -> None:
        """Fold a line into the child's open turn. Events are consumed, not emitted:
        the live stream is the parent timeline's; fanning child events into it would
        interleave two timelines under one label."""
        turn = tail.open_turn
        if turn is None:
            raise RuntimeError(f"subagent {tail.agent_id}: no open turn to fold into")
        written = len(turn.accumulator.messages)
        for _ in turn.accumulator.consume(line):
            pass
        stamp(turn.accumulator.messages[written:], line.timestamp)
        turn.end_offset = end
        turn.last_line_uuid = line.uuid

    async def close_subagent_turn(self, state: TailState, tail: SubagentTail) -> None:
        """Commit the child's open turn as a child run: keyed `<agentId>:<promptId>`
        (which subagent, which parent turn — a bare `promptId` would collide with the
        parent run it belongs to), linked by `parent_run_id` = that `promptId` and the
        harvested spawning call. A subagent has no human prompt or answer, so it never
        touches the ledger."""
        turn = tail.open_turn
        tail.open_turn = None
        if turn is None:
            return
        run_id = f"{tail.agent_id}:{turn.prompt_id}"
        if run_id in tail.recorded or not turn.accumulator.messages:
            return
        conversation = tail.conversation
        if conversation is None:
            raise RuntimeError(f"subagent {tail.agent_id} has no conversation")
        with claude_logfire.span(
            "claude.tailer.commit_subagent_turn {run_id} [{session_id}]",
            run_id=run_id,
            session_id=state.session_id,
            start_offset=turn.start_offset,
            end_offset=turn.end_offset,
            messages=len(turn.accumulator.messages),
        ) as span:
            async with self.locks.hold(
                state.session_id, acquire_timeout=COMMIT_LOCK_TIMEOUT
            ):
                run = await self.conversation_manager.record_external_run(
                    conversation,
                    run_id=run_id,
                    messages=turn.accumulator.messages,
                    name=CLAUDE_NATIVE_ID,
                    model_name=turn.accumulator.model_name,
                    permission_mode=turn.permission_mode,
                    cwd=Path(turn.cwd) if turn.cwd else None,
                    native_session_id=tail.agent_id,
                    source=turn.source,
                    start_offset=turn.start_offset,
                    end_offset=turn.end_offset,
                    last_line_uuid=turn.last_line_uuid,
                    parent_run_id=turn.prompt_id,
                    parent_tool_call_id=state.subagent_calls.get(
                        (tail.agent_id, turn.prompt_id)
                    ),
                )
                span.set_attribute("committed", run is not None)
                if run is None:
                    return
                tail.recorded.add(run_id)
                logger.info(
                    "session %s: subagent %s turn %s synced — %d messages, bytes %d-%d",
                    state.session_id,
                    tail.agent_id,
                    turn.prompt_id,
                    len(turn.accumulator.messages),
                    turn.start_offset,
                    turn.end_offset,
                )

    async def close_subagent_turns(self, state: TailState) -> None:
        """Commit every child's trailing open turn — the subagent counterpart of the
        final `close_turn`, at a session-ending drain (`finish_remote`)."""
        for tail in state.subagents.values():
            await self.close_subagent_turn(state, tail)

conversation_manager instance-attribute 

conversation_manager = conversation_manager

thread_manager instance-attribute 

thread_manager = thread_manager

locks instance-attribute 

locks = locks if locks is not None else SessionLocks()

sessions instance-attribute 

sessions: dict[str, TailState] = {}

stop_turn async 

stop_turn(session_id: str, prompt_id: str | None) -> None

The turn just stopped — its Stop hook fired: drain it now instead of waiting for the next prompt line to close it.

The stream route relays finalize on stop_event, the client ships its final lines and answers eof, and finish_remote commits the stopped turn — the tail process exits, and the next prompt's launcher spawns a fresh one. Scoped by prompt_id so nothing newer than the stopped turn ever commits early; without one there is nothing to scope, and the turn keeps its old boundary. A session no stream covers has nothing to drain.

Source code in octomate/tentacles/claude/tailer.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
async def stop_turn(self, session_id: str, prompt_id: str | None) -> None:
    """The turn just stopped — its `Stop` hook fired: drain it now instead of
    waiting for the next prompt line to close it.

    The stream route relays `finalize` on `stop_event`, the client ships its
    final lines and answers `eof`, and `finish_remote` commits the stopped turn —
    the tail process exits, and the next prompt's launcher spawns a fresh one.
    Scoped by `prompt_id` so nothing newer than the stopped turn ever commits
    early; without one there is nothing to scope, and the turn keeps its old
    boundary. A session no stream covers has nothing to drain.
    """
    if prompt_id is None:
        return
    state = self.sessions.get(session_id)
    if state is None:
        return
    state.drain_turn = prompt_id
    state.stop_event.set()

finalize async 

finalize(session_id: str) -> None

End a session (SessionEnd): relay the drain to its stream — the route watches stop_event and sends the client finalize; the client drains, answers eof, and finish_remote commits the trailing turns and sets drained. The wait is bounded: a client that is gone leaves its turns for the next connect, and the SessionEnd hook must not hang on it. A session no stream covers has nothing to finalize.

Source code in octomate/tentacles/claude/tailer.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
async def finalize(self, session_id: str) -> None:
    """End a session (`SessionEnd`): relay the drain to its stream — the route
    watches `stop_event` and sends the client `finalize`; the client drains,
    answers `eof`, and `finish_remote` commits the trailing turns and sets
    `drained`. The wait is bounded: a client that is gone leaves its turns for
    the next connect, and the SessionEnd hook must not hang on it. A session no
    stream covers has nothing to finalize."""
    state = self.sessions.get(session_id)
    if state is None:
        return
    state.stop_event.set()
    try:
        await asyncio.wait_for(state.drained.wait(), REMOTE_DRAIN_TIMEOUT)
    except TimeoutError:
        logger.warning(
            "session %s: remote tail did not drain within %ss; its trailing "
            "turns are left for the next connect to recover",
            session_id,
            REMOTE_DRAIN_TIMEOUT,
        )

shutdown async 

shutdown() -> None
Source code in octomate/tentacles/claude/tailer.py 
232
233
async def shutdown(self) -> None:
    self.sessions.clear()

attach_remote async 

attach_remote(
    session_id: str,
    transcript_path: Path,
    sender: UserProfile,
    address: ChannelAddress,
    impulses: ObjectSendStream[Impulse],
) -> tuple[TailState, dict[str, int]]

Register a streamed session and answer where each of its files resumes.

The offsets are recomputed from the committed runs, per file: the session transcript under SESSION_FILE, each known subagent under its agent id — so the client holds no durable cursor and a reconnect is just re-asking. A lingering registration (its client died without a close) is replaced; the dead route feeds the state object it attached, and its commits are idempotent.

Source code in octomate/tentacles/claude/tailer.py 
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
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
async def attach_remote(
    self,
    session_id: str,
    transcript_path: Path,
    sender: UserProfile,
    address: ChannelAddress,
    impulses: ObjectSendStream[Impulse],
) -> tuple[TailState, dict[str, int]]:
    """Register a streamed session and answer where each of its files resumes.

    The offsets are recomputed from the committed runs, per file: the session
    transcript under `SESSION_FILE`, each known subagent
    under its agent id — so the client holds no durable cursor and a reconnect is
    just re-asking. A lingering registration (its client died without a close) is
    replaced; the dead route feeds the state object it attached, and its commits
    are idempotent.
    """
    state = TailState(
        session_id=session_id,
        transcript_path=transcript_path,
        stop_event=asyncio.Event(),
        sender=sender,
        address=address,
        impulses=impulses,
    )
    await self.prepare(state)

    conversation = state.conversation
    if conversation is None:
        raise RuntimeError(f"session {session_id} prepared without a conversation")
    state.offset = max(
        (
            run.end_offset or 0
            for run in conversation.runs
            if isinstance(run, ExternalAgentRun)
        ),
        default=0,
    )
    offsets: dict[str, int] = {SESSION_FILE: state.offset}
    for child in await self.conversation_manager.subagents(conversation.id):
        if not child.subagent_id:
            continue
        tail = self.subagent_tail(state, child.subagent_id)
        tail.conversation = child
        tail.recorded = child.native_run_ids
        tail.offset = max(
            (
                run.end_offset or 0
                for run in child.runs
                if isinstance(run, ExternalAgentRun)
            ),
            default=0,
        )
        offsets[child.subagent_id] = tail.offset
    self.sessions[session_id] = state
    logger.info(
        "session %s: remote tail attached, resuming at %s", session_id, offsets
    )
    return state, offsets

feed_remote async 

feed_remote(
    state: TailState,
    agent_id: str | None,
    raw: str,
    start: int,
    end: int,
) -> None

Advance a remote session by one framed line — the pumps' per-line body, against offsets the client measured in its own file. Takes the state rather than a session id so a superseded connection keeps feeding the state it attached, never whoever registered after it.

Source code in octomate/tentacles/claude/tailer.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
async def feed_remote(
    self, state: TailState, agent_id: str | None, raw: str, start: int, end: int
) -> None:
    """Advance a remote session by one framed line — the pumps' per-line body,
    against offsets the client measured in its own file. Takes the state rather
    than a session id so a superseded connection keeps feeding the state it
    attached, never whoever registered after it."""
    if agent_id is None:
        state.offset = end
        line = self.parse_line(raw, state.session_id)
        if line is not None:
            await self.process_line(state, line, start, end)
        return
    tail = self.subagent_tail(state, agent_id)
    if tail.conversation is None:
        await self.prepare_subagent(state, tail)
    tail.offset = end
    line = self.parse_line(raw, state.session_id)
    if line is not None:
        await self.process_subagent_line(state, tail, line, start, end)

finish_remote async 

finish_remote(state: TailState) -> None

A remote session's clean end — the client drained to EOF and said eof.

After the session's own end (finalize relayed at SessionEnd, or the idle drain), commit everything still open. After a relayed Stop (drain_turn set), commit only the stopped turn: a prompt queued into the drain window is left open to re-stream whole on the next connect, and subagent turns keep their own boundary — SubagentStop commits them, and force-closing here would truncate a background child still running across turns. Either way, release any finalize waiter and reclaim the registry slot.

Source code in octomate/tentacles/claude/tailer.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
async def finish_remote(self, state: TailState) -> None:
    """A remote session's clean end — the client drained to EOF and said `eof`.

    After the session's own end (`finalize` relayed at `SessionEnd`, or the idle
    drain), commit everything still open. After a relayed `Stop` (`drain_turn`
    set), commit only the stopped turn: a prompt queued into the drain window is
    left open to re-stream whole on the next connect, and subagent turns keep
    their own boundary — `SubagentStop` commits them, and force-closing here
    would truncate a background child still running across turns. Either way,
    release any finalize waiter and reclaim the registry slot."""
    try:
        if state.drain_turn is None:
            await self.close_turn(state)
            await self.close_subagent_turns(state)
        else:
            turn = state.open_turn
            if turn is not None and turn.prompt_id == state.drain_turn:
                await self.close_turn(state)
    finally:
        state.drained.set()
        self.detach_remote(state)

detach_remote 

detach_remote(state: TailState) -> None

Drop a remote session's registration without committing anything: the connection died mid-stream, so its open turns' bytes were never provably complete. The next connect resumes from the committed offsets and re-streams them.

Source code in octomate/tentacles/claude/tailer.py 
338
339
340
341
342
343
344
def detach_remote(self, state: TailState) -> None:
    """Drop a remote session's registration without committing anything: the
    connection died mid-stream, so its open turns' bytes were never provably
    complete. The next connect resumes from the committed offsets and re-streams
    them."""
    if self.sessions.get(state.session_id) is state:
        del self.sessions[state.session_id]

prepare async 

prepare(state: TailState) -> None

Resolve the session's conversation and seed the committed-turn guard from it, so a resume never re-commits a run the last run already wrote.

Source code in octomate/tentacles/claude/tailer.py 
346
347
348
349
350
async def prepare(self, state: TailState) -> None:
    """Resolve the session's conversation and seed the committed-turn guard from it,
    so a resume never re-commits a run the last run already wrote."""
    state.conversation = await self.ensure_session(state.session_id)
    state.recorded = state.conversation.native_run_ids

ensure_session async 

ensure_session(session_id: str) -> Conversation

Map a native Claude session id to its Octomate home — the session's thread and the conversation hanging off it — creating either if this is its first sighting.

Source code in octomate/tentacles/claude/tailer.py 
352
353
354
355
356
357
358
359
360
async def ensure_session(self, session_id: str) -> Conversation:
    """Map a native Claude session id to its Octomate home — the session's thread and
    the conversation hanging off it — creating either if this is its first sighting."""
    thread = await self.thread_manager.ensure(
        ThreadKey(CLAUDE_NATIVE_ID, "thread", session_id)
    )
    return await self.conversation_manager.ensure(
        thread.id, agent_tentacle_id=CLAUDE_NATIVE_ID
    )

parse_line staticmethod 

parse_line(
    raw: str | bytes, session_id: str
) -> TranscriptLine | None

One typed line, or None for a blank / malformed / unmodeled one — skipped but never allowed to wedge ingest; the caller advances its cursor either way.

Source code in octomate/tentacles/claude/tailer.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
@staticmethod
def parse_line(raw: str | bytes, session_id: str) -> TranscriptLine | None:
    """One typed line, or None for a blank / malformed / unmodeled one — skipped
    but never allowed to wedge ingest; the caller advances its cursor either way."""
    stripped = raw.strip()
    if not stripped:
        return None
    try:
        return transcript_line_adapter.validate_json(stripped)
    except ValidationError:
        logger.debug(
            "skipping unmodeled/malformed transcript line in session %s",
            session_id,
        )
        return None

process_line async 

process_line(
    state: TailState,
    line: TranscriptLine,
    start: int,
    end: int,
) -> None

Route one typed line: a prompt_source user line opens a turn (closing the previous one); other user (tool-result) and assistant lines fold into the open turn, and a stop_hook_summary closes it. Inline is_sidechain lines are skipped — transcripts since 2.1.177 keep subagents in their own files (pump_subagents), so this guards only against an older transcript's inline relics.

Source code in octomate/tentacles/claude/tailer.py 
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
async def process_line(
    self, state: TailState, line: TranscriptLine, start: int, end: int
) -> None:
    """Route one typed line: a `prompt_source` user line opens a turn (closing the
    previous one); other user (tool-result) and assistant lines fold into the open
    turn, and a `stop_hook_summary` closes it. Inline `is_sidechain` lines are
    skipped — transcripts since 2.1.177 keep subagents in their own files
    (`pump_subagents`), so this guards only against an older transcript's inline
    relics."""
    if isinstance(line, TranscriptUserLine):
        if line.is_sidechain:
            return
        await self.harvest_subagent_call(state, line)
        if line.prompt_source is not None:
            await self.close_turn(state)
            await self.record_permission_mode(state, line.permission_mode)
            state.open_turn = await self.begin_turn(state, line, start, end)
        elif state.open_turn is not None:
            await self.fold(state, line, end)
    elif isinstance(line, TranscriptAssistantLine):
        if line.is_sidechain:
            return
        if state.open_turn is not None:
            await self.fold(state, line, end)
    elif isinstance(line, TranscriptSystemLine):
        # Claude writes it once the turn's `Stop` hooks have run, delivered or
        # not: a server down through the stop never hears the hook.
        if line.subtype == "stop_hook_summary" and not line.is_sidechain:
            await self.close_turn(state)
    elif isinstance(line, TranscriptAiTitleLine):
        await self.record_title(state, line.ai_title)

record_title async 

record_title(state: TailState, title: str) -> None

Carry the name Claude gave this session onto the rows that show it.

The transcript restates the title on every turn and revises it as the work turns out to be about something else, so the comparison — not the write — is what runs per line. The thread takes it too: a native thread is its session, and a name about the work beats the line the thread opened with.

Source code in octomate/tentacles/claude/tailer.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
async def record_title(self, state: TailState, title: str) -> None:
    """Carry the name Claude gave this session onto the rows that show it.

    The transcript restates the title on every turn and revises it as the work
    turns out to be about something else, so the comparison — not the write —
    is what runs per line. The thread takes it too: a native thread is its
    session, and a name about the work beats the line the thread opened with.
    """
    conversation = state.conversation
    if conversation is None or conversation.name == title.strip():
        return
    await self.conversation_manager.set_name(conversation, title)
    thread = await self.thread_manager.ensure(
        ThreadKey(CLAUDE_NATIVE_ID, "thread", state.session_id)
    )
    await self.thread_manager.rename(thread, title)

harvest_subagent_call async 

harvest_subagent_call(
    state: TailState, line: TranscriptUserLine
) -> None

Remember which parent tool call spawned which subagent: an Agent call's tool-result line names the child (toolUseResult.agentId) while its content block names the call. Stamps child runs' parent_tool_call_id at commit.

Source code in octomate/tentacles/claude/tailer.py 
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
453
454
455
456
457
458
async def harvest_subagent_call(
    self, state: TailState, line: TranscriptUserLine
) -> None:
    """Remember which parent tool call spawned which subagent: an `Agent` call's
    tool-result line names the child (`toolUseResult.agentId`) while its content
    block names the call. Stamps child runs' `parent_tool_call_id` at commit."""
    result = line.tool_use_result
    if not isinstance(result, dict):
        return
    agent_id = result.get("agentId")
    if not isinstance(agent_id, str) or not agent_id:
        return
    content = line.message.content
    if not isinstance(content, list):
        return
    tool_use_id = next(
        (
            block.get("tool_use_id")
            for block in content
            if isinstance(block, dict) and block.get("type") == "tool_result"
        ),
        None,
    )
    if not isinstance(tool_use_id, str):
        return
    async with self.locks.hold(state.session_id):
        state.subagent_calls[agent_id, line.prompt_id] = tool_use_id
        await self.conversation_manager.link_parent_run(
            f"{agent_id}:{line.prompt_id}",
            parent_run_id=line.prompt_id,
            parent_tool_call_id=tool_use_id,
        )

fold async 

fold(
    state: TailState,
    line: TranscriptAssistantLine | TranscriptUserLine,
    end: int,
) -> None

Fold a line into the open turn: consume it (pushing its live events) and extend the turn's byte range and provenance to cover it.

Source code in octomate/tentacles/claude/tailer.py 
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
async def fold(
    self,
    state: TailState,
    line: TranscriptAssistantLine | TranscriptUserLine,
    end: int,
) -> None:
    """Fold a line into the open turn: consume it (pushing its live events) and
    extend the turn's byte range and provenance to cover it."""
    turn = state.open_turn
    if turn is None:
        raise RuntimeError(f"session {state.session_id}: no open turn to fold into")
    if turn.native is None:
        return
    written = len(turn.accumulator.messages)
    if isinstance(line, TranscriptUserLine) and (text := prompt_text(line.message)):
        await turn.native.prompt(
            message_id=line.uuid,
            text=text,
            timestamp=line.timestamp,
        )
    await turn.native.fold(turn.accumulator.consume(line))
    stamp(turn.accumulator.messages[written:], line.timestamp)
    turn.end_offset = end
    turn.last_line_uuid = line.uuid

record_permission_mode async 

record_permission_mode(
    state: TailState, permission_mode: str | None
) -> None

Keep the conversation's posture at what the session's own transcript says.

Every prompt line carries the mode that turn ran under, so a ⇧⇥ in the client reaches Octomate on the operator's next message. Observed, never set: nothing here can change a running client's mode, and the console shows it read-only.

A mode Claude has that this build's vocabulary does not is skipped rather than stored — the column is validated on read, so an unknown one would take the conversation out of circulation instead of merely being unrecognized.

Source code in octomate/tentacles/claude/tailer.py 
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
async def record_permission_mode(
    self, state: TailState, permission_mode: str | None
) -> None:
    """Keep the conversation's posture at what the session's own transcript says.

    Every prompt line carries the mode that turn ran under, so a ⇧⇥ in the client
    reaches Octomate on the operator's next message. Observed, never set: nothing
    here can change a running client's mode, and the console shows it read-only.

    A mode Claude has that this build's vocabulary does not is skipped rather than
    stored — the column is validated on read, so an unknown one would take the
    conversation out of circulation instead of merely being unrecognized.
    """
    conversation = state.conversation
    if conversation is None or permission_mode is None:
        return
    if permission_mode == conversation.permission_mode:
        return
    if not is_claude_mode(permission_mode):
        logger.debug(
            "session %s reports permission mode %r, which this build does not "
            "model; leaving the conversation at %r",
            state.session_id,
            permission_mode,
            conversation.permission_mode,
        )
        return
    await self.conversation_manager.set_permission_mode(
        conversation, permission_mode
    )

begin_turn async 

begin_turn(
    state: TailState,
    line: TranscriptUserLine,
    start: int,
    end: int,
) -> OpenTurn

Open the native lifecycle at the transcript's prompt boundary.

Source code in octomate/tentacles/claude/tailer.py 
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
async def begin_turn(
    self, state: TailState, line: TranscriptUserLine, start: int, end: int
) -> OpenTurn:
    """Open the native lifecycle at the transcript's prompt boundary."""
    accumulator = ClaudeRunAccumulator()
    native = await NativeTurn.open(
        run_id=line.prompt_id,
        recorded=state.recorded,
        conversations=self.conversation_manager,
        threads=self.thread_manager,
        locks=self.locks,
        address=state.address,
        sender=state.sender,
        impulses=state.impulses,
        messages=accumulator.messages,
    )
    text = prompt_text(line.message)
    if native is not None:
        await native.prompt(
            message_id=line.prompt_id, text=text, timestamp=line.timestamp
        )
    return OpenTurn(
        prompt_id=line.prompt_id,
        source=line.entrypoint,
        cwd=line.cwd,
        start_offset=start,
        end_offset=end,
        accumulator=accumulator,
        last_line_uuid=line.uuid,
        permission_mode=line.permission_mode,
        native=native,
    )

close_turn async 

close_turn(state: TailState) -> None

Commit the open turn as an ExternalAgentRun with its byte range, then bind the turn's human-ledger rows to it. Idempotent by prompt_id, so a re-driven overlap after a resume is a safe no-op. Held under the session lock so the commit cannot race a hook reserving the same turn.

A commit that cannot be made propagates instead of being skipped past. A reconnect resumes from the last committed turn's end_offset, which only points at the right bytes while the committed turns are the earliest ones: skipping a turn and streaming on would let a later turn push that mark past the gap, stranding the skipped turn where no re-stream could reach it. Failing keeps the mark honest — the next connect re-streams the turn from there and commits it.

Source code in octomate/tentacles/claude/tailer.py 
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
async def close_turn(self, state: TailState) -> None:
    """Commit the open turn as an `ExternalAgentRun` with its byte range, then bind
    the turn's human-ledger rows to it. Idempotent by `prompt_id`, so a re-driven
    overlap after a resume is a safe no-op. Held under the session lock so the commit
    cannot race a hook reserving the same turn.

    A commit that cannot be made propagates instead of being skipped past. A
    reconnect resumes from the last committed turn's `end_offset`, which only
    points at the right bytes while the committed turns are the *earliest* ones:
    skipping a turn and streaming on would let a later turn push that mark past
    the gap, stranding the skipped turn where no re-stream could reach it.
    Failing keeps the mark honest — the next connect re-streams the turn from
    there and commits it.
    """
    turn = state.open_turn
    state.open_turn = None
    if turn is None or turn.native is None:
        return
    conversation = state.conversation
    if conversation is None:
        raise RuntimeError(f"session {state.session_id} has no conversation")
    with claude_logfire.span(
        "claude.tailer.commit_turn {prompt_id} [{session_id}]",
        prompt_id=turn.prompt_id,
        session_id=state.session_id,
        start_offset=turn.start_offset,
        end_offset=turn.end_offset,
        messages=len(turn.accumulator.messages),
    ) as span:
        async with self.locks.hold(
            state.session_id, acquire_timeout=COMMIT_LOCK_TIMEOUT
        ):
            run = await self.conversation_manager.record_external_run(
                conversation,
                run_id=turn.prompt_id,
                messages=turn.accumulator.messages,
                name=CLAUDE_NATIVE_ID,
                model_name=turn.accumulator.model_name,
                permission_mode=turn.permission_mode,
                cwd=Path(turn.cwd) if turn.cwd else None,
                native_session_id=state.session_id,
                source=turn.source,
                start_offset=turn.start_offset,
                end_offset=turn.end_offset,
                last_line_uuid=turn.last_line_uuid,
            )
            span.set_attribute("committed", run is not None)
            await turn.native.close(
                run, run_id=turn.prompt_id, answer=turn.accumulator.result_text
            )
            logger.info(
                "session %s: turn %s synced — %d messages, bytes %d-%d",
                state.session_id,
                turn.prompt_id,
                len(turn.accumulator.messages),
                turn.start_offset,
                turn.end_offset,
            )

subagent_tail staticmethod 

subagent_tail(
    state: TailState, agent_id: str
) -> SubagentTail
Source code in octomate/tentacles/claude/tailer.py 
608
609
610
611
612
613
614
615
616
617
@staticmethod
def subagent_tail(state: TailState, agent_id: str) -> SubagentTail:
    tail = state.subagents.get(agent_id)
    if tail is None:
        tail = SubagentTail(
            agent_id=agent_id,
            path=state.subagents_dir / f"agent-{agent_id}.jsonl",
        )
        state.subagents[agent_id] = tail
    return tail

finish_subagent async 

finish_subagent(
    session_id: str,
    agent_id: str,
    *,
    final_answer: str | None = None,
) -> None

Drain and commit one subagent on its SubagentStop — the child's finalize, which a Claude child needs: its turns frame on promptId change, so nothing in its own file closes the last one. The hook is synchronous but the transcript writer is not: the final answer line races it (measured: usually loses by ~1-2KB). The child's lines arrive through feed_remote on their own; when the event names the answer, wait until the stream yields it, bounded so a mismatch can never hang the pipe — tail.offset says whether the writer has gone quiet. A session no stream covers has nothing to settle.

Source code in octomate/tentacles/claude/tailer.py 
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
async def finish_subagent(
    self, session_id: str, agent_id: str, *, final_answer: str | None = None
) -> None:
    """Drain and commit one subagent on its `SubagentStop` — the child's
    `finalize`, which a Claude child needs: its turns frame on promptId change,
    so nothing in its own file closes the last one. The hook is synchronous but
    the transcript writer is not: the final answer line races it (measured:
    usually loses by ~1-2KB). The child's lines arrive through `feed_remote` on
    their own; when the event names the answer, wait until the stream yields it,
    bounded so a mismatch can never hang the pipe — `tail.offset` says whether
    the writer has gone quiet. A session no stream covers has nothing to
    settle."""
    state = self.sessions.get(session_id)
    if state is None:
        return
    if state.conversation is None:
        await self.prepare(state)
    tail = self.subagent_tail(state, agent_id)
    # The turn this stop is for. If a resume arrives inside the settle window,
    # the stream itself closes this turn on the promptId change and opens the
    # next — which must NOT be committed by this stop, mid-flight.
    target = tail.open_turn
    deadline = monotonic() + SUBAGENT_SETTLE_TIMEOUT
    quiet = 0
    while not self.subagent_settled(tail, final_answer):
        if quiet >= 2:
            break  # writer went silent: the file is as final as it gets
        if monotonic() >= deadline:
            logger.warning(
                "session %s: subagent %s kept writing past the settle window "
                "without yielding its announced answer; committing what it "
                "reached",
                session_id,
                agent_id,
            )
            break
        before = tail.offset
        await asyncio.sleep(SUBAGENT_SETTLE_POLL)
        quiet = quiet + 1 if tail.offset == before else 0
    if target is None or tail.open_turn is target:
        await self.close_subagent_turn(state, tail)

subagent_settled staticmethod 

subagent_settled(
    tail: SubagentTail, final_answer: str | None
) -> bool

The fast path: the drained turn already carries the exact answer the stop event announced. Divergence (a truncated or differently-joined hook payload — no contract promises byte equality) is not an error: the caller's byte-quiescence fallback settles those, so strictness here only ever costs one extra poll beat, never a wrong commit.

Source code in octomate/tentacles/claude/tailer.py 
661
662
663
664
665
666
667
668
669
670
671
672
673
@staticmethod
def subagent_settled(tail: SubagentTail, final_answer: str | None) -> bool:
    """The fast path: the drained turn already carries the exact answer the
    stop event announced. Divergence (a truncated or differently-joined hook
    payload — no contract promises byte equality) is not an error: the caller's
    byte-quiescence fallback settles those, so strictness here only ever costs
    one extra poll beat, never a wrong commit."""
    if final_answer is None:
        return False  # nothing announced; quiescence decides
    turn = tail.open_turn
    if turn is None:
        return False
    return turn.accumulator.result_text.strip() == final_answer.strip()

prepare_subagent async 

prepare_subagent(
    state: TailState, tail: SubagentTail
) -> None

Resolve the child's conversation under the session's thread and seed its committed-turn guard — the first time this child's lines are fed.

Source code in octomate/tentacles/claude/tailer.py 
675
676
677
678
679
680
681
682
683
684
685
686
687
async def prepare_subagent(self, state: TailState, tail: SubagentTail) -> None:
    """Resolve the child's conversation under the session's thread and seed its
    committed-turn guard — the first time this child's lines are fed."""
    parent = state.conversation
    if parent is None:
        raise RuntimeError(f"session {state.session_id} has no conversation")
    tail.conversation = await self.conversation_manager.ensure(
        parent.thread_id,
        agent_tentacle_id=CLAUDE_NATIVE_ID,
        subagent_id=tail.agent_id,
        parent_conversation_id=parent.id,
    )
    tail.recorded = tail.conversation.native_run_ids

process_subagent_line async 

process_subagent_line(
    state: TailState,
    tail: SubagentTail,
    line: TranscriptLine,
    start: int,
    end: int,
) -> None

Frame a subagent's turns on promptId change — the only marker its file carries (promptSource is absent, and every line is is_sidechain, so neither parent rule applies). Each turn's promptId is the parent turn that drove it; a resumed subagent opens its next turn on whatever user line arrives first, which is measured to be a tool result, not a prompt.

Source code in octomate/tentacles/claude/tailer.py 
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
async def process_subagent_line(
    self,
    state: TailState,
    tail: SubagentTail,
    line: TranscriptLine,
    start: int,
    end: int,
) -> None:
    """Frame a subagent's turns on `promptId` change — the only marker its file
    carries (`promptSource` is absent, and every line is `is_sidechain`, so
    neither parent rule applies). Each turn's `promptId` is the *parent* turn
    that drove it; a resumed subagent opens its next turn on whatever user line
    arrives first, which is measured to be a tool result, not a prompt."""
    if isinstance(line, TranscriptUserLine):
        turn = tail.open_turn
        if turn is None or line.prompt_id != turn.prompt_id:
            await self.close_subagent_turn(state, tail)
            tail.open_turn = self.begin_subagent_turn(line, start, end)
        else:
            self.fold_subagent(tail, line, end)
    elif isinstance(line, TranscriptAssistantLine):
        if tail.open_turn is not None:
            self.fold_subagent(tail, line, end)

begin_subagent_turn 

begin_subagent_turn(
    line: TranscriptUserLine, start: int, end: int
) -> OpenTurn
Source code in octomate/tentacles/claude/tailer.py 
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
def begin_subagent_turn(
    self, line: TranscriptUserLine, start: int, end: int
) -> OpenTurn:
    accumulator = ClaudeRunAccumulator()
    text = prompt_text(line.message)
    if text:
        accumulator.begin(text)
    # The opener may itself carry tool results (a resumed subagent's turn opens on
    # one); consuming is a no-op for a plain prompt line.
    for _ in accumulator.consume(line):
        pass
    stamp(accumulator.messages, line.timestamp)
    return OpenTurn(
        prompt_id=line.prompt_id,
        source=line.entrypoint,
        cwd=line.cwd,
        start_offset=start,
        end_offset=end,
        accumulator=accumulator,
        last_line_uuid=line.uuid,
        permission_mode=line.permission_mode,
    )

fold_subagent 

fold_subagent(
    tail: SubagentTail,
    line: TranscriptAssistantLine | TranscriptUserLine,
    end: int,
) -> None

Fold a line into the child's open turn. Events are consumed, not emitted: the live stream is the parent timeline's; fanning child events into it would interleave two timelines under one label.

Source code in octomate/tentacles/claude/tailer.py 
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
def fold_subagent(
    self,
    tail: SubagentTail,
    line: TranscriptAssistantLine | TranscriptUserLine,
    end: int,
) -> None:
    """Fold a line into the child's open turn. Events are consumed, not emitted:
    the live stream is the parent timeline's; fanning child events into it would
    interleave two timelines under one label."""
    turn = tail.open_turn
    if turn is None:
        raise RuntimeError(f"subagent {tail.agent_id}: no open turn to fold into")
    written = len(turn.accumulator.messages)
    for _ in turn.accumulator.consume(line):
        pass
    stamp(turn.accumulator.messages[written:], line.timestamp)
    turn.end_offset = end
    turn.last_line_uuid = line.uuid

close_subagent_turn async 

close_subagent_turn(
    state: TailState, tail: SubagentTail
) -> None

Commit the child's open turn as a child run: keyed <agentId>:<promptId> (which subagent, which parent turn — a bare promptId would collide with the parent run it belongs to), linked by parent_run_id = that promptId and the harvested spawning call. A subagent has no human prompt or answer, so it never touches the ledger.

Source code in octomate/tentacles/claude/tailer.py 
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
async def close_subagent_turn(self, state: TailState, tail: SubagentTail) -> None:
    """Commit the child's open turn as a child run: keyed `<agentId>:<promptId>`
    (which subagent, which parent turn — a bare `promptId` would collide with the
    parent run it belongs to), linked by `parent_run_id` = that `promptId` and the
    harvested spawning call. A subagent has no human prompt or answer, so it never
    touches the ledger."""
    turn = tail.open_turn
    tail.open_turn = None
    if turn is None:
        return
    run_id = f"{tail.agent_id}:{turn.prompt_id}"
    if run_id in tail.recorded or not turn.accumulator.messages:
        return
    conversation = tail.conversation
    if conversation is None:
        raise RuntimeError(f"subagent {tail.agent_id} has no conversation")
    with claude_logfire.span(
        "claude.tailer.commit_subagent_turn {run_id} [{session_id}]",
        run_id=run_id,
        session_id=state.session_id,
        start_offset=turn.start_offset,
        end_offset=turn.end_offset,
        messages=len(turn.accumulator.messages),
    ) as span:
        async with self.locks.hold(
            state.session_id, acquire_timeout=COMMIT_LOCK_TIMEOUT
        ):
            run = await self.conversation_manager.record_external_run(
                conversation,
                run_id=run_id,
                messages=turn.accumulator.messages,
                name=CLAUDE_NATIVE_ID,
                model_name=turn.accumulator.model_name,
                permission_mode=turn.permission_mode,
                cwd=Path(turn.cwd) if turn.cwd else None,
                native_session_id=tail.agent_id,
                source=turn.source,
                start_offset=turn.start_offset,
                end_offset=turn.end_offset,
                last_line_uuid=turn.last_line_uuid,
                parent_run_id=turn.prompt_id,
                parent_tool_call_id=state.subagent_calls.get(
                    (tail.agent_id, turn.prompt_id)
                ),
            )
            span.set_attribute("committed", run is not None)
            if run is None:
                return
            tail.recorded.add(run_id)
            logger.info(
                "session %s: subagent %s turn %s synced — %d messages, bytes %d-%d",
                state.session_id,
                tail.agent_id,
                turn.prompt_id,
                len(turn.accumulator.messages),
                turn.start_offset,
                turn.end_offset,
            )

close_subagent_turns async 

close_subagent_turns(state: TailState) -> None

Commit every child's trailing open turn — the subagent counterpart of the final close_turn, at a session-ending drain (finish_remote).

Source code in octomate/tentacles/claude/tailer.py 
814
815
816
817
818
async def close_subagent_turns(self, state: TailState) -> None:
    """Commit every child's trailing open turn — the subagent counterpart of the
    final `close_turn`, at a session-ending drain (`finish_remote`)."""
    for tail in state.subagents.values():
        await self.close_subagent_turn(state, tail)

stamp 

stamp(
    messages: list[ModelMessage], timestamp: datetime | None
) -> None

Date the messages a transcript record produced by that record's own clock — the accumulator stamps now, which is wrong for a run replayed off disk (AgentRun.started_at and displayed times read from these).

Source code in octomate/tentacles/claude/tailer.py 
67
68
69
70
71
72
73
74
def stamp(messages: list[PydanticModelMessage], timestamp: datetime | None) -> None:
    """Date the messages a transcript record produced by that record's own clock — the
    accumulator stamps `now`, which is wrong for a run replayed off disk
    (`AgentRun.started_at` and displayed times read from these)."""
    if timestamp is None:
        return
    for message in messages:
        message.timestamp = timestamp

octomate.tentacles.deepseek.tailer

Assembly of native dsh turns from a streamed session log.

History entries streamed in by octomate deepseek tail feed DeepseekEventTailer, which commits each completed turn as an ExternalAgentRun.

logger module-attribute 

logger = logging.getLogger(__name__)

COMMIT_LOCK_TIMEOUT module-attribute 

COMMIT_LOCK_TIMEOUT = 30.0

STOP_SETTLE_TIMEOUT module-attribute 

STOP_SETTLE_TIMEOUT = 5.0

PendingPrompt dataclass

One human user/message — the ledger row's raw material.

Source code in octomate/tentacles/deepseek/tailer.py 
90
91
92
93
94
95
96
97
@dataclass
class PendingPrompt:
    """One human `user/message` — the ledger row's raw material."""

    text: str
    time: datetime
    seq: int
    via_gateway: bool

text instance-attribute 

text: str

time instance-attribute 

time: datetime

seq instance-attribute 

seq: int

via_gateway instance-attribute 

via_gateway: bool

OpenTurn dataclass

The turn currently being assembled off the stream — its accumulator fed entry by entry, its seq range advancing to cover the last one folded in.

Source code in octomate/tentacles/deepseek/tailer.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@dataclass
class OpenTurn:
    """The turn currently being assembled off the stream — its accumulator fed
    entry by entry, its seq range advancing to cover the last one folded in."""

    turn: int | None
    start_seq: int
    end_seq: int
    # The prompt arrived through the /api gateway (it carries the gateway's
    # rpcId) rather than being typed into a local CLI/TUI.
    prompt_via_gateway: bool
    permission_mode: str | None  # Observed permission preset for this turn.
    accumulator: DeepseekRunAccumulator = field(default_factory=DeepseekRunAccumulator)
    native: NativeTurn | None = None

turn instance-attribute 

turn: int | None

start_seq instance-attribute 

start_seq: int

end_seq instance-attribute 

end_seq: int

prompt_via_gateway instance-attribute 

prompt_via_gateway: bool

permission_mode instance-attribute 

permission_mode: str | None

accumulator class-attribute instance-attribute 

accumulator: DeepseekRunAccumulator = field(
    default_factory=DeepseekRunAccumulator
)

native class-attribute instance-attribute 

native: NativeTurn | None = None

TailState dataclass

One streamed session. Its log lives on the client machine that streams it (octomate deepseek tail, reading that machine's dsh gateway); the stream route feeds feed_remote, which drives the per-event assembly. transcript_path is the client's claim in the client's own namespace, a label never opened here.

Source code in octomate/tentacles/deepseek/tailer.py 
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
@dataclass
class TailState:
    """One streamed session. Its log lives on the client machine that streams
    it (`octomate deepseek tail`, reading that machine's dsh gateway); the
    stream route feeds `feed_remote`, which drives the per-event assembly.
    `transcript_path` is the client's claim in the client's own namespace, a
    label never opened here."""

    session_id: str
    transcript_path: Path
    cwd: str
    stop_event: asyncio.Event
    # The verified bearer's own profile — who every ledger row this stream
    # writes is attributed to.
    sender: UserProfile
    address: ChannelAddress  # The native surface resolved by the tentacle.
    impulses: ObjectSendStream[Impulse]  # The publisher open for this connection.
    conversation: Conversation | None = None
    recorded: set[str] = field(default_factory=set)
    floor: int = -1
    prompts: list[PendingPrompt] = field(default_factory=list)
    open_turn: OpenTurn | None = None
    # Pulses on every committed turn; a relayed `Stop` clears it and waits so
    # the finalize goes out only once the stopped turn is durable (or the
    # bounded wait ran out).
    commit_pulse: asyncio.Event = field(default_factory=asyncio.Event)

session_id instance-attribute 

session_id: str

transcript_path instance-attribute 

transcript_path: Path

cwd instance-attribute 

cwd: str

stop_event instance-attribute 

stop_event: Event

sender instance-attribute 

sender: UserProfile

address instance-attribute 

address: ChannelAddress

impulses instance-attribute 

impulses: ObjectSendStream[Impulse]

conversation class-attribute instance-attribute 

conversation: Conversation | None = None

recorded class-attribute instance-attribute 

recorded: set[str] = field(default_factory=set)

floor class-attribute instance-attribute 

floor: int = -1

prompts class-attribute instance-attribute 

prompts: list[PendingPrompt] = field(default_factory=list)

open_turn class-attribute instance-attribute 

open_turn: OpenTurn | None = None

commit_pulse class-attribute instance-attribute 

commit_pulse: Event = field(default_factory=asyncio.Event)

DeepseekEventTailer

Assembles native dsh turns from history entries streamed in by octomate deepseek tail — the stream is the only assembler: the server never opens a session log and never speaks to the client machine's dsh.

The client reads its local gateway (session.history, which serves the zstd-framed log decoded and unpacked, cold sessions included) and ships each entry as one framed line, the event's dense seq standing where a file tail's byte offsets stand. Entries fold through the same DeepseekRunAccumulator the driven tentacle uses; a turn commits on its own turn/end line as an ExternalAgentRun keyed <session>:<turn>, carrying the seq range it was assembled from — the committed floor a reconnecting client is told to resume past. An open turn never commits at a connection boundary: the next connect re-streams it whole.

Source code in octomate/tentacles/deepseek/tailer.py 
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
271
272
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
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
394
395
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
437
438
439
440
441
442
443
444
445
446
447
class DeepseekEventTailer:
    """Assembles native dsh turns from history entries streamed in by
    `octomate deepseek tail` — the stream is the only assembler: the server
    never opens a session log and never speaks to the client machine's dsh.

    The client reads its local gateway (`session.history`, which serves the
    zstd-framed log decoded and unpacked, cold sessions included) and ships
    each entry as one framed line, the event's dense `seq` standing where a
    file tail's byte offsets stand. Entries fold through the same
    `DeepseekRunAccumulator` the driven tentacle uses; a turn commits on its
    own `turn/end` line as an `ExternalAgentRun` keyed `<session>:<turn>`,
    carrying the seq range it was assembled from — the committed floor a
    reconnecting client is told to resume past. An open turn never commits at
    a connection boundary: the next connect re-streams it whole.
    """

    def __init__(
        self,
        conversation_manager: ConversationManager,
        thread_manager: ThreadManager,
        projects: ProjectManager | None = None,
        locks: SessionLocks | None = None,
    ) -> None:
        self.conversation_manager = conversation_manager
        self.thread_manager = thread_manager
        self.projects = projects if projects is not None else ProjectManager()
        self.locks = locks if locks is not None else SessionLocks()
        self.sessions: dict[str, TailState] = {}

    async def attach_remote(
        self,
        session_id: str,
        transcript_path: Path,
        cwd: str,
        sender: UserProfile,
        address: ChannelAddress,
        impulses: ObjectSendStream[Impulse],
    ) -> tuple[TailState, dict[str, int]]:
        """Register a streamed session and answer where it resumes: the seq
        after the committed floor, recomputed from the durable runs — the
        client holds no cursor of its own. A lingering registration (its
        client died without a close) is replaced; the dead route keeps feeding
        the state it attached, and its commits are idempotent."""
        state = TailState(
            session_id, transcript_path, cwd, asyncio.Event(), sender, address, impulses
        )
        thread = await self.session_thread(session_id, cwd)
        state.conversation = await self.conversation_manager.ensure(
            thread.id, agent_tentacle_id=DEEPSEEK_NATIVE_ID
        )
        state.recorded = state.conversation.native_run_ids
        state.floor = committed_floor(state.conversation)
        self.sessions[session_id] = state
        logger.info("session %s: remote tail attached", session_id)
        return state, {SESSION_FILE: state.floor + 1}

    async def feed_remote(self, state: TailState, raw: str) -> None:
        """Fold one history entry; the route validates the sequence framing."""
        try:
            entry = history_entry_adapter.validate_json(raw)
        except ValidationError:
            logger.debug("skipping unparseable dsh history entry")
            return
        await self.process_entry(state, entry)

    def detach_remote(self, state: TailState) -> None:
        """Drop a streamed session's registration. Nothing commits on the way
        out — a dsh turn closes on its own `turn/end` line, so a turn still
        open here is mid-flight, and the next connect re-streams it whole from
        the committed floor."""
        if self.sessions.get(state.session_id) is state:
            del self.sessions[state.session_id]

    async def stop_turn(self, session_id: str) -> None:
        """The turn just stopped — its `Stop` hook fired: wait for the stopped
        turn's streamed close to commit, then ask the drain. Runs detached
        from the hook (dsh's stopping seam waits on the hook serially, ahead
        of the very `turn/end` this waits for). A session no stream covers has
        nothing to settle: its turns land at the next connect."""
        state = self.sessions.get(session_id)
        if state is None:
            return
        state.commit_pulse.clear()
        with contextlib.suppress(TimeoutError):
            await asyncio.wait_for(state.commit_pulse.wait(), STOP_SETTLE_TIMEOUT)
        state.stop_event.set()

    async def shutdown(self) -> None:
        self.sessions.clear()

    async def session_thread(self, session_id: str, cwd: str) -> Thread:
        """This session's thread, filed under the project already holding the
        directory it runs in — and under none when no project holds it. As on
        the Claude side, a dsh session is not evidence of a project, so
        nothing is registered here; the cwd is recorded on each run
        regardless."""
        holder = self.projects.resolve(Path(cwd)) if cwd else None
        project = self.projects.get(holder) if holder is not None else None
        return await self.thread_manager.ensure(
            ThreadKey(DEEPSEEK_NATIVE_ID, "thread", session_id),
            project=project,
        )

    async def process_entry(self, state: TailState, entry: HistoryEntry) -> None:
        event = entry.event
        if event.seq <= state.floor:
            return
        if event.type == "session/title":
            try:
                name = SessionTitle.model_validate(event.data).title
            except ValidationError:
                return
            if not name or not name.strip():
                return
            conversation = state.conversation
            assert conversation is not None
            async with self.locks.hold(state.session_id):
                await self.conversation_manager.set_name(conversation, name)
                thread = await self.session_thread(state.session_id, state.cwd)
                await self.thread_manager.rename(thread, name)
            return
        if event.type == "permission/preset":
            preset = permission_preset_of(event)
            await self.record_permission_mode(state, preset)
            if state.open_turn is not None and preset is not None:
                state.open_turn.permission_mode = preset
            return
        if state.open_turn is None:
            if event.type == "user/message":
                prompt = human_prompt(event)
                if prompt is not None:
                    state.prompts.append(prompt)
            elif event.type == "turn/start":
                prompts = state.prompts
                state.prompts = []
                number = (
                    event.data.get("turn") if isinstance(event.data, dict) else None
                )
                number = number if isinstance(number, int) else None
                run_id = (
                    f"{state.session_id}:{number if number is not None else event.seq}"
                )
                state.open_turn = OpenTurn(
                    turn=number,
                    start_seq=prompts[0].seq if prompts else event.seq,
                    end_seq=event.seq,
                    prompt_via_gateway=any(prompt.via_gateway for prompt in prompts),
                    permission_mode=(
                        state.conversation.permission_mode
                        if state.conversation is not None
                        else None
                    ),
                )
                state.open_turn.native = await NativeTurn.open(
                    run_id=run_id,
                    recorded=state.recorded,
                    conversations=self.conversation_manager,
                    threads=self.thread_manager,
                    locks=self.locks,
                    address=state.address,
                    sender=state.sender,
                    impulses=state.impulses,
                    messages=state.open_turn.accumulator.messages,
                )
                if state.open_turn.native is None:
                    return
                for prompt in prompts:
                    await self.record_prompt(state, state.open_turn, prompt)
                await self.fold(state.open_turn, state, entry)
            return
        turn = state.open_turn
        if turn.native is None:
            if event.type == "turn/end":
                state.open_turn = None
            return
        if event.type == "user/message":
            prompt = human_prompt(event)
            if prompt is not None:
                await self.record_prompt(state, turn, prompt)
                turn.prompt_via_gateway = turn.prompt_via_gateway or prompt.via_gateway
        await self.fold(turn, state, entry)
        if turn.accumulator.turn_ended:
            state.open_turn = None
            end = turn_end_of(event)
            if end is not None and end.turn is not None:
                turn.turn = end.turn
            await self.commit_turn(state, turn)

    async def record_prompt(
        self, state: TailState, turn: OpenTurn, prompt: PendingPrompt
    ) -> None:
        """Keep each human input where it occurs, including mid-turn steering.

        The event seq is stable across reconnects, even before dsh reports a turn
        number. Its ledger row grants access while the turn is still streaming.
        """
        assert turn.native is not None
        await turn.native.prompt(
            message_id=f"{state.session_id}:message:{prompt.seq}",
            text=prompt.text,
            timestamp=prompt.time,
            metadata=deepseek_metadata(),
        )

    async def fold(self, turn: OpenTurn, state: TailState, entry: HistoryEntry) -> None:
        """Fold and publish one entry, dating messages by the log's clock.

        Committed seqs never reach the fold. The native thread receives only run
        brackets, with the end delivered after its ledger binding commits.
        Re-streaming an open turn starts it again without waking a reflex.
        """
        accumulator = turn.accumulator
        written = len(accumulator.messages)
        frame = SessionEventFrame(
            type="session/event",
            session_id=state.session_id,
            event=entry.event,
            view=entry.view,
        )
        assert turn.native is not None
        await turn.native.fold(accumulator.consume(frame))
        stamp(accumulator.messages[written:], event_time(entry.event))
        turn.end_seq = entry.event.seq

    async def commit_turn(self, state: TailState, turn: OpenTurn) -> None:
        session_id = state.session_id
        run_key = turn.turn if turn.turn is not None else turn.start_seq
        run_id = f"{session_id}:{run_key}"
        if turn.native is None:
            return
        messages = turn.accumulator.messages
        conversation = state.conversation
        assert conversation is not None
        with deepseek_logfire.span(
            "deepseek.tailer.commit_turn {run_id} [{session_id}]",
            run_id=run_id,
            session_id=session_id,
            start_seq=turn.start_seq,
            end_seq=turn.end_seq,
            messages=len(messages),
        ) as span:
            async with self.locks.hold(session_id, acquire_timeout=COMMIT_LOCK_TIMEOUT):
                run = await self.conversation_manager.record_external_run(
                    conversation,
                    run_id=run_id,
                    messages=messages,
                    name=DEEPSEEK_NATIVE_ID,
                    model_name=(
                        turn.accumulator.route.model
                        if turn.accumulator.route is not None
                        else None
                    ),
                    permission_mode=turn.permission_mode,
                    cwd=Path(state.cwd) if state.cwd else None,
                    native_session_id=session_id,
                    source="gateway" if turn.prompt_via_gateway else "local",
                    start_offset=turn.start_seq,
                    end_offset=turn.end_seq,
                )
                span.set_attribute("committed", run is not None)
                await turn.native.close(
                    run, run_id=run_id, answer=turn.accumulator.result_text
                )
                state.floor = max(state.floor, turn.end_seq)
                state.commit_pulse.set()
                logger.info(
                    "session %s: turn %s synced — %d messages, seqs %d-%d",
                    session_id,
                    run_id,
                    len(messages),
                    turn.start_seq,
                    turn.end_seq,
                )

    async def record_permission_mode(
        self, state: TailState, preset: str | None
    ) -> None:
        """Keep the conversation's posture at what the session log last said.
        Observed names are retained even when absent from the driven catalog."""
        conversation = state.conversation
        if conversation is None or preset is None:
            return
        if preset == conversation.permission_mode:
            return
        await self.conversation_manager.set_permission_mode(conversation, preset)

conversation_manager instance-attribute 

conversation_manager = conversation_manager

thread_manager instance-attribute 

thread_manager = thread_manager

projects instance-attribute 

projects = (
    projects if projects is not None else ProjectManager()
)

locks instance-attribute 

locks = locks if locks is not None else SessionLocks()

sessions instance-attribute 

sessions: dict[str, TailState] = {}

attach_remote async 

attach_remote(
    session_id: str,
    transcript_path: Path,
    cwd: str,
    sender: UserProfile,
    address: ChannelAddress,
    impulses: ObjectSendStream[Impulse],
) -> tuple[TailState, dict[str, int]]

Register a streamed session and answer where it resumes: the seq after the committed floor, recomputed from the durable runs — the client holds no cursor of its own. A lingering registration (its client died without a close) is replaced; the dead route keeps feeding the state it attached, and its commits are idempotent.

Source code in octomate/tentacles/deepseek/tailer.py 
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
async def attach_remote(
    self,
    session_id: str,
    transcript_path: Path,
    cwd: str,
    sender: UserProfile,
    address: ChannelAddress,
    impulses: ObjectSendStream[Impulse],
) -> tuple[TailState, dict[str, int]]:
    """Register a streamed session and answer where it resumes: the seq
    after the committed floor, recomputed from the durable runs — the
    client holds no cursor of its own. A lingering registration (its
    client died without a close) is replaced; the dead route keeps feeding
    the state it attached, and its commits are idempotent."""
    state = TailState(
        session_id, transcript_path, cwd, asyncio.Event(), sender, address, impulses
    )
    thread = await self.session_thread(session_id, cwd)
    state.conversation = await self.conversation_manager.ensure(
        thread.id, agent_tentacle_id=DEEPSEEK_NATIVE_ID
    )
    state.recorded = state.conversation.native_run_ids
    state.floor = committed_floor(state.conversation)
    self.sessions[session_id] = state
    logger.info("session %s: remote tail attached", session_id)
    return state, {SESSION_FILE: state.floor + 1}

feed_remote async 

feed_remote(state: TailState, raw: str) -> None

Fold one history entry; the route validates the sequence framing.

Source code in octomate/tentacles/deepseek/tailer.py 
219
220
221
222
223
224
225
226
async def feed_remote(self, state: TailState, raw: str) -> None:
    """Fold one history entry; the route validates the sequence framing."""
    try:
        entry = history_entry_adapter.validate_json(raw)
    except ValidationError:
        logger.debug("skipping unparseable dsh history entry")
        return
    await self.process_entry(state, entry)

detach_remote 

detach_remote(state: TailState) -> None

Drop a streamed session's registration. Nothing commits on the way out — a dsh turn closes on its own turn/end line, so a turn still open here is mid-flight, and the next connect re-streams it whole from the committed floor.

Source code in octomate/tentacles/deepseek/tailer.py 
228
229
230
231
232
233
234
def detach_remote(self, state: TailState) -> None:
    """Drop a streamed session's registration. Nothing commits on the way
    out — a dsh turn closes on its own `turn/end` line, so a turn still
    open here is mid-flight, and the next connect re-streams it whole from
    the committed floor."""
    if self.sessions.get(state.session_id) is state:
        del self.sessions[state.session_id]

stop_turn async 

stop_turn(session_id: str) -> None

The turn just stopped — its Stop hook fired: wait for the stopped turn's streamed close to commit, then ask the drain. Runs detached from the hook (dsh's stopping seam waits on the hook serially, ahead of the very turn/end this waits for). A session no stream covers has nothing to settle: its turns land at the next connect.

Source code in octomate/tentacles/deepseek/tailer.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
async def stop_turn(self, session_id: str) -> None:
    """The turn just stopped — its `Stop` hook fired: wait for the stopped
    turn's streamed close to commit, then ask the drain. Runs detached
    from the hook (dsh's stopping seam waits on the hook serially, ahead
    of the very `turn/end` this waits for). A session no stream covers has
    nothing to settle: its turns land at the next connect."""
    state = self.sessions.get(session_id)
    if state is None:
        return
    state.commit_pulse.clear()
    with contextlib.suppress(TimeoutError):
        await asyncio.wait_for(state.commit_pulse.wait(), STOP_SETTLE_TIMEOUT)
    state.stop_event.set()

shutdown async 

shutdown() -> None
Source code in octomate/tentacles/deepseek/tailer.py 
250
251
async def shutdown(self) -> None:
    self.sessions.clear()

session_thread async 

session_thread(session_id: str, cwd: str) -> Thread

This session's thread, filed under the project already holding the directory it runs in — and under none when no project holds it. As on the Claude side, a dsh session is not evidence of a project, so nothing is registered here; the cwd is recorded on each run regardless.

Source code in octomate/tentacles/deepseek/tailer.py 
253
254
255
256
257
258
259
260
261
262
263
264
async def session_thread(self, session_id: str, cwd: str) -> Thread:
    """This session's thread, filed under the project already holding the
    directory it runs in — and under none when no project holds it. As on
    the Claude side, a dsh session is not evidence of a project, so
    nothing is registered here; the cwd is recorded on each run
    regardless."""
    holder = self.projects.resolve(Path(cwd)) if cwd else None
    project = self.projects.get(holder) if holder is not None else None
    return await self.thread_manager.ensure(
        ThreadKey(DEEPSEEK_NATIVE_ID, "thread", session_id),
        project=project,
    )

process_entry async 

process_entry(
    state: TailState, entry: HistoryEntry
) -> None
Source code in octomate/tentacles/deepseek/tailer.py 
266
267
268
269
270
271
272
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
async def process_entry(self, state: TailState, entry: HistoryEntry) -> None:
    event = entry.event
    if event.seq <= state.floor:
        return
    if event.type == "session/title":
        try:
            name = SessionTitle.model_validate(event.data).title
        except ValidationError:
            return
        if not name or not name.strip():
            return
        conversation = state.conversation
        assert conversation is not None
        async with self.locks.hold(state.session_id):
            await self.conversation_manager.set_name(conversation, name)
            thread = await self.session_thread(state.session_id, state.cwd)
            await self.thread_manager.rename(thread, name)
        return
    if event.type == "permission/preset":
        preset = permission_preset_of(event)
        await self.record_permission_mode(state, preset)
        if state.open_turn is not None and preset is not None:
            state.open_turn.permission_mode = preset
        return
    if state.open_turn is None:
        if event.type == "user/message":
            prompt = human_prompt(event)
            if prompt is not None:
                state.prompts.append(prompt)
        elif event.type == "turn/start":
            prompts = state.prompts
            state.prompts = []
            number = (
                event.data.get("turn") if isinstance(event.data, dict) else None
            )
            number = number if isinstance(number, int) else None
            run_id = (
                f"{state.session_id}:{number if number is not None else event.seq}"
            )
            state.open_turn = OpenTurn(
                turn=number,
                start_seq=prompts[0].seq if prompts else event.seq,
                end_seq=event.seq,
                prompt_via_gateway=any(prompt.via_gateway for prompt in prompts),
                permission_mode=(
                    state.conversation.permission_mode
                    if state.conversation is not None
                    else None
                ),
            )
            state.open_turn.native = await NativeTurn.open(
                run_id=run_id,
                recorded=state.recorded,
                conversations=self.conversation_manager,
                threads=self.thread_manager,
                locks=self.locks,
                address=state.address,
                sender=state.sender,
                impulses=state.impulses,
                messages=state.open_turn.accumulator.messages,
            )
            if state.open_turn.native is None:
                return
            for prompt in prompts:
                await self.record_prompt(state, state.open_turn, prompt)
            await self.fold(state.open_turn, state, entry)
        return
    turn = state.open_turn
    if turn.native is None:
        if event.type == "turn/end":
            state.open_turn = None
        return
    if event.type == "user/message":
        prompt = human_prompt(event)
        if prompt is not None:
            await self.record_prompt(state, turn, prompt)
            turn.prompt_via_gateway = turn.prompt_via_gateway or prompt.via_gateway
    await self.fold(turn, state, entry)
    if turn.accumulator.turn_ended:
        state.open_turn = None
        end = turn_end_of(event)
        if end is not None and end.turn is not None:
            turn.turn = end.turn
        await self.commit_turn(state, turn)

record_prompt async 

record_prompt(
    state: TailState, turn: OpenTurn, prompt: PendingPrompt
) -> None

Keep each human input where it occurs, including mid-turn steering.

The event seq is stable across reconnects, even before dsh reports a turn number. Its ledger row grants access while the turn is still streaming.

Source code in octomate/tentacles/deepseek/tailer.py 
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
async def record_prompt(
    self, state: TailState, turn: OpenTurn, prompt: PendingPrompt
) -> None:
    """Keep each human input where it occurs, including mid-turn steering.

    The event seq is stable across reconnects, even before dsh reports a turn
    number. Its ledger row grants access while the turn is still streaming.
    """
    assert turn.native is not None
    await turn.native.prompt(
        message_id=f"{state.session_id}:message:{prompt.seq}",
        text=prompt.text,
        timestamp=prompt.time,
        metadata=deepseek_metadata(),
    )

fold async 

fold(
    turn: OpenTurn, state: TailState, entry: HistoryEntry
) -> None

Fold and publish one entry, dating messages by the log's clock.

Committed seqs never reach the fold. The native thread receives only run brackets, with the end delivered after its ledger binding commits. Re-streaming an open turn starts it again without waking a reflex.

Source code in octomate/tentacles/deepseek/tailer.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
async def fold(self, turn: OpenTurn, state: TailState, entry: HistoryEntry) -> None:
    """Fold and publish one entry, dating messages by the log's clock.

    Committed seqs never reach the fold. The native thread receives only run
    brackets, with the end delivered after its ledger binding commits.
    Re-streaming an open turn starts it again without waking a reflex.
    """
    accumulator = turn.accumulator
    written = len(accumulator.messages)
    frame = SessionEventFrame(
        type="session/event",
        session_id=state.session_id,
        event=entry.event,
        view=entry.view,
    )
    assert turn.native is not None
    await turn.native.fold(accumulator.consume(frame))
    stamp(accumulator.messages[written:], event_time(entry.event))
    turn.end_seq = entry.event.seq

commit_turn async 

commit_turn(state: TailState, turn: OpenTurn) -> None
Source code in octomate/tentacles/deepseek/tailer.py 
387
388
389
390
391
392
393
394
395
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
async def commit_turn(self, state: TailState, turn: OpenTurn) -> None:
    session_id = state.session_id
    run_key = turn.turn if turn.turn is not None else turn.start_seq
    run_id = f"{session_id}:{run_key}"
    if turn.native is None:
        return
    messages = turn.accumulator.messages
    conversation = state.conversation
    assert conversation is not None
    with deepseek_logfire.span(
        "deepseek.tailer.commit_turn {run_id} [{session_id}]",
        run_id=run_id,
        session_id=session_id,
        start_seq=turn.start_seq,
        end_seq=turn.end_seq,
        messages=len(messages),
    ) as span:
        async with self.locks.hold(session_id, acquire_timeout=COMMIT_LOCK_TIMEOUT):
            run = await self.conversation_manager.record_external_run(
                conversation,
                run_id=run_id,
                messages=messages,
                name=DEEPSEEK_NATIVE_ID,
                model_name=(
                    turn.accumulator.route.model
                    if turn.accumulator.route is not None
                    else None
                ),
                permission_mode=turn.permission_mode,
                cwd=Path(state.cwd) if state.cwd else None,
                native_session_id=session_id,
                source="gateway" if turn.prompt_via_gateway else "local",
                start_offset=turn.start_seq,
                end_offset=turn.end_seq,
            )
            span.set_attribute("committed", run is not None)
            await turn.native.close(
                run, run_id=run_id, answer=turn.accumulator.result_text
            )
            state.floor = max(state.floor, turn.end_seq)
            state.commit_pulse.set()
            logger.info(
                "session %s: turn %s synced — %d messages, seqs %d-%d",
                session_id,
                run_id,
                len(messages),
                turn.start_seq,
                turn.end_seq,
            )

record_permission_mode async 

record_permission_mode(
    state: TailState, preset: str | None
) -> None

Keep the conversation's posture at what the session log last said. Observed names are retained even when absent from the driven catalog.

Source code in octomate/tentacles/deepseek/tailer.py 
437
438
439
440
441
442
443
444
445
446
447
async def record_permission_mode(
    self, state: TailState, preset: str | None
) -> None:
    """Keep the conversation's posture at what the session log last said.
    Observed names are retained even when absent from the driven catalog."""
    conversation = state.conversation
    if conversation is None or preset is None:
        return
    if preset == conversation.permission_mode:
        return
    await self.conversation_manager.set_permission_mode(conversation, preset)

committed_floor 

committed_floor(conversation: Conversation) -> int

The highest event seq any committed turn covers — the stream resumes at the seq after it.

Source code in octomate/tentacles/deepseek/tailer.py 
64
65
66
67
68
69
70
71
72
73
74
def committed_floor(conversation: Conversation) -> int:
    """The highest event seq any committed turn covers — the stream resumes at
    the seq after it."""
    return max(
        (
            run.end_offset
            for run in conversation.runs
            if isinstance(run, ExternalAgentRun) and run.end_offset is not None
        ),
        default=-1,
    )

event_time 

event_time(event: SessionEvent) -> datetime

A session event's own clock — time is epoch milliseconds.

Source code in octomate/tentacles/deepseek/tailer.py 
77
78
79
def event_time(event: SessionEvent) -> datetime:
    """A session event's own clock — `time` is epoch milliseconds."""
    return datetime.fromtimestamp(event.time / 1000, tz=UTC)

stamp 

stamp(
    messages: list[ModelMessage], timestamp: datetime
) -> None

Date the messages an event produced by that event's own clock — the accumulator stamps now, which is wrong for a turn replayed off the log (AgentRun.started_at and ordering both read from these).

Source code in octomate/tentacles/deepseek/tailer.py 
82
83
84
85
86
87
def stamp(messages: list[ModelMessage], timestamp: datetime) -> None:
    """Date the messages an event produced by that event's own clock — the
    accumulator stamps `now`, which is wrong for a turn replayed off the log
    (`AgentRun.started_at` and ordering both read from these)."""
    for message in messages:
        message.timestamp = timestamp

human_prompt 

human_prompt(event: SessionEvent) -> PendingPrompt | None

The human's own words in a user/message, or None. dsh logs injected user-role messages in the same turn — agent-instructions, plugin runtime context — which are the harness talking, not the ledger's prompt; human prompts and late question replies are the person's input.

Source code in octomate/tentacles/deepseek/tailer.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def human_prompt(event: SessionEvent) -> PendingPrompt | None:
    """The human's own words in a `user/message`, or None. dsh logs injected
    user-role messages in the same turn — `agent-instructions`, `plugin`
    runtime context — which are the harness talking, not the ledger's prompt;
    human prompts and late question replies are the person's input."""
    message = user_message_of(event)
    if message is None or message.source.kind not in {"user", "user-question-reply"}:
        return None
    text = text_of(message.content)
    if not text:
        return None
    return PendingPrompt(
        text=text,
        time=event_time(event),
        seq=event.seq,
        via_gateway=message.source.rpc_id is not None,
    )