Method create_session

src/server/session_manager.py:58–100

Allocate a new session ID and record it in STARTING state. Raises ``RuntimeError`` if ``max_sessions`` would be exceeded. Caller is responsible for actually spawning the agent subprocess and then calling ``attach_process`` + ``mark_running``.

(
        self,
        *,
        cwd: str | None = None,
        permission_mode: str | None = None,
    )

    # ─── Mutators ──────────────────────────────────────────────────────

    def create_session(
        self,
        *,
        cwd: str | None = None,
        permission_mode: str | None = None,
    ) -> SessionInfo:
        """Allocate a new session ID and record it in STARTING state.

        Raises ``RuntimeError`` if ``max_sessions`` would be exceeded.
        Caller is responsible for actually spawning the agent
        subprocess and then calling ``attach_process`` + ``mark_running``.
        """
        if self.max_sessions is not None and self._active_count() >= self.max_sessions:
            raise RuntimeError(
                f'Direct Connect server: max_sessions ({self.max_sessions}) reached'
            )
        sid = f'ds_{_uuid.uuid4().hex}'
        now = time.time()
        info = SessionInfo(
            id=sid,
            status=SessionState.STARTING,
            created_at=now,
            work_dir=cwd or self.workspace,
            last_active_at=now,
        )
        self._sessions[sid] = info
        # Persist so a server restart can resume — even before the
        # subprocess actually starts. The transcript_session_id
        # initially equals the session_id; if the subprocess uses a
        # different transcript ID, the server can update via
        # ``update_transcript_id`` (out of scope for this minimal cut).
        add_entry(
            SessionIndexEntry(
                session_id=sid,
                transcript_session_id=sid,
                cwd=info.work_dir,
                created_at=info.created_at,
                last_active_at=info.created_at,
                permission_mode=permission_mode,
            ),
            path=self.index_path,
        )
        return info

    def attach_process(
        self,