Skip to main content
Claw Code tracks conversation state across turns through a layered system: QueryEnginePort holds live in-memory state, TranscriptStore maintains an ordered entry log, and StoredSession serialises that state to disk between runs.

​
QueryEnginePort session fields

QueryEnginePort (in query_engine.py) is the stateful centre of a session: 
@dataclass
class QueryEnginePort:
    manifest: PortManifest
    config: QueryEngineConfig
    session_id: str                          # uuid4 hex, auto-generated
    mutable_messages: list[str]              # growing list of submitted prompts
    permission_denials: list[PermissionDenial]
    total_usage: UsageSummary
    transcript_store: TranscriptStore
A new engine is created with QueryEnginePort.from_workspace(). An existing session is restored with QueryEnginePort.from_saved_session(session_id).

​
QueryEngineConfig

QueryEngineConfig controls turn-loop behaviour: 
@dataclass(frozen=True)
class QueryEngineConfig:
    max_turns: int = 8             # hard cap on turns before stopping
    max_budget_tokens: int = 2000  # combined token budget (input + output)
    compact_after_turns: int = 12  # trim message list above this length
    structured_output: bool = False
    structured_retry_limit: int = 2
When len(mutable_messages) >= max_turns, submit_message() returns immediately with stop_reason='max_turns_reached' without processing the prompt.
After each turn, projected token usage is compared against this limit. If input_tokens + output_tokens would exceed the budget, the turn is recorded but stop_reason is set to 'max_budget_reached'.
When mutable_messages grows beyond compact_after_turns, it is trimmed to the most recent compact_after_turns entries. The TranscriptStore is compacted by the same amount.
When True, submit_message() returns JSON instead of plain text. Failed serialisation is retried up to structured_retry_limit times before raising RuntimeError.

​
TurnResult

Every call to submit_message() returns a TurnResult: 
@dataclass(frozen=True)
class TurnResult:
    prompt: str
    output: str
    matched_commands: tuple[str, ...]
    matched_tools: tuple[str, ...]
    permission_denials: tuple[PermissionDenial, ...]
    usage: UsageSummary
    stop_reason: str

​
Stop reasons

ValueMeaning
"completed"Turn processed normally; budget and turn limits not exceeded.
"max_turns_reached"mutable_messages length hit config.max_turns before the prompt was processed.
"max_budget_reached"Projected token usage after this turn would exceed config.max_budget_tokens.
In run_turn_loop(), the loop exits early on any stop reason other than "completed": 
for turn in range(max_turns):
    result = engine.submit_message(turn_prompt, ...)
    results.append(result)
    if result.stop_reason != 'completed':
        break

​
UsageSummary

UsageSummary accumulates token counts across turns. It is immutable — each call to add_turn() returns a new instance: 
@dataclass(frozen=True)
class UsageSummary:
    input_tokens: int = 0
    output_tokens: int = 0

    def add_turn(self, prompt: str, output: str) -> 'UsageSummary':
        return UsageSummary(
            input_tokens=self.input_tokens + len(prompt.split()),
            output_tokens=self.output_tokens + len(output.split()),
        )
Token counts are currently estimated from word counts (len(text.split())), not from a tokeniser. They serve as a lightweight budget proxy during porting; replace add_turn() with a real tokeniser for production use.

​
TranscriptStore

TranscriptStore (in transcript.py) is a lightweight ordered log: 
@dataclass
class TranscriptStore:
    entries: list[str]
    flushed: bool = False
MethodBehaviour
append(entry)Adds a string entry; sets flushed=False.
compact(keep_last)Trims entries to the most recent keep_last items.
replay()Returns all entries as an immutable tuple.
flush()Sets flushed=True without clearing entries.
The engine calls compact_messages_if_needed() after each turn, which delegates compaction to both mutable_messages and the transcript store. flush() is called automatically by persist_session() before writing to disk.

​
StoredSession and disk persistence

StoredSession (in session_store.py) is the serialisation envelope: 
@dataclass(frozen=True)
class StoredSession:
    session_id: str
    messages: tuple[str, ...]
    input_tokens: int
    output_tokens: int
Sessions are written to .port_sessions/<session_id>.json: 
# Save
path = save_session(StoredSession(session_id=..., messages=..., ...))

# Load
stored = load_session(session_id)
engine = QueryEnginePort.from_saved_session(session_id)
The default directory is .port_sessions/ relative to the working directory. Pass a custom directory argument to either function to override this. 
# Flush and persist a session from the CLI
python3 -m src.main flush-transcript "my prompt"

# Reload a previously saved session
python3 -m src.main load-session <session_id>

​
HistoryLog

HistoryLog (in history.py) is a simple runtime event journal attached to a RuntimeSession: 
@dataclass(frozen=True)
class HistoryEvent:
    title: str
    detail: str

@dataclass
class HistoryLog:
    events: list[HistoryEvent]

    def add(self, title: str, detail: str) -> None: ...
    def as_markdown(self) -> str: ...
PortRuntime.bootstrap_session() records events at each stage of the session lifecycle: 
history.add('context', f'python_files={context.python_file_count}, archive_available={context.archive_available}')
history.add('registry', f'commands={len(PORTED_COMMANDS)}, tools={len(PORTED_TOOLS)}')
history.add('routing', f'matches={len(matches)} for prompt={prompt!r}')
history.add('execution', f'command_execs={len(command_execs)} tool_execs={len(tool_execs)}')
history.add('turn', f'commands=... tools=... denials=... stop={turn_result.stop_reason}')
history.add('session_store', persisted_session_path)
The full history is included in the Markdown output of RuntimeSession.as_markdown().

Build docs developers (and LLMs) love

Get started for freeTalk to us