ss-tools/.opencode/skills/molecular-cot-logging/SKILL.md

name, description

name	description
molecular-cot-logging	Structured logging protocol for agent-driven development, based on molecular Long CoT bonds (REASON/REFLECT/EXPLORE). Replaces legacy Entry/Exit/Coherence markers. Python + Svelte.

#region MolecularCoTLogging [C:5] [TYPE Skill] [SEMANTICS reasoning,runtime,logging,agentic] @BRIEF Structured logging protocol for agent-driven development, based on molecular Long CoT bonds (Deep-Reasoning, Self-Reflection, Self-Exploration). Replaces legacy Entry/Exit/Coherence markers. @RELATION DEPENDS_ON -> [Std.Semantics.Core] @RELATION DISPATCHES -> [Std.Semantics.Python] @RELATION DISPATCHES -> [Std.Semantics.Svelte] @RATIONALE Long CoT chains need stabilisation through explicit reasoning bonds. The three-marker system (REASON/REFLECT/EXPLORE) maps directly to the molecular CoT paper and produces machine-readable execution traces that LLM agents can parse, analyse, and use for fine-tuning (MoLE-Syn bond distributions). @REJECTED Legacy Entry/Exit/Action/Coherence markers rejected — they are too generic, do not map to reasoning structure, and prevent traceability graph analysis. Plain-text logging rejected — JSON lines are mandatory for agent parsing. @DATA_CONTRACT LogEntry -> { ts: str, level: str, trace_id: str, span_id?: str, src: str, marker: REASON|REFLECT|EXPLORE, intent: str, payload?: object, error?: str } @INVARIANT Every log line MUST carry exactly one valid marker (REASON | REFLECT | EXPLORE). No markerless log lines in C4/C5 code. @INVARIANT trace_id MUST propagate via ContextVar across async boundaries. Every incoming request or background job seeds a new trace_id.

Purpose

Enable transparent agent-driven development by producing machine-readable execution traces that directly reflect the reasoning structure of the code. Every log line becomes an edge in a traceability graph that an LLM agent can parse, analyse, and optionally use for fine-tuning (via MoLE-Syn-like bond distributions).

Core principles (from the Molecular CoT paper)

Long CoT chains are stabilised by three "chemical bonds":

Bond	Marker	Function
Deep-Reasoning	REASON	Extends the logical backbone
Self-Reflection	REFLECT	Folds back to validate or correct previous steps
Self-Exploration	EXPLORE	Branches into alternatives when an assumption fails

Our logs annotate every semantically meaningful step with exactly one of these markers.

I. Log Entry Specification

Every log record MUST be a JSON object on a single line with the following keys:

Field	Required	Type	Description
ts	yes	string	ISO-8601 timestamp with millisecond precision
level	yes	string	Standard log level (INFO, DEBUG, WARNING, ERROR)
trace_id	yes	string	UUID of the incoming HTTP request or background job
span_id	no	string	UUID of the current function/block scope (optional)
src	yes	string	Qualified function name, e.g. AuthRepository.get_user_by_username
marker	yes	string	One of REASON, REFLECT, EXPLORE (see below)
intent	yes	string	Human-readable one-line description of what this step intends to do/verify
payload	no	object	Arbitrary key-value data relevant to the step (params, result snippet)
error	conditional	string	Error message or reason. Optional for REASON/REFLECT, required for EXPLORE markers when a fallback or violation is taken

Example

{"ts":"2026-05-12T14:31:39.577","level":"INFO","trace_id":"d874a1b2-...","span_id":"...","src":"AuthRepository.get_user_by_username","marker":"REASON","intent":"Fetch user by username","payload":{"username":"admin"}}

II. Semantic Marker Usage

REASON (Deep-Reasoning)

• When: BEFORE an operation that extends the logical chain (DB query, API call, computation).
• Level: INFO by default, DEBUG for high-frequency loops.
• intent: Describes what the code is about to do.
• payload: Input parameters, context values.
• Effect: This is the primary "deep-reasoning" step that forms the backbone of the trace.

log("AuthRepository.get_user_by_username", "REASON",
    "Fetch user by username", {"username": username})

REFLECT (Self-Reflection)

• When: AFTER an operation to verify the outcome or check invariants.
• Level: INFO on success, WARNING if invariants partially degrade.
• intent: Describes what is being verified.
• payload: Result summary, status codes, row counts.
• Effect: Folds the logical chain back on itself — the agent sees cause + effect in two adjacent lines.

log("AuthRepository.get_user_by_username", "REFLECT",
    "User found", {"found": user is not None, "user_id": user.id if user else None})

EXPLORE (Self-Exploration)

• When: An expected condition is violated and the code enters a fallback, error handler, or alternative path.
• Level: WARNING for recoverable fallbacks, ERROR for unrecoverable failures.
• intent: Describes what assumption failed.
• payload: Relevant state at the branch point.
• error: Required. Explain what assumption was violated.
• Effect: Creates a branch in the trace — a future agent can see why the happy path was not taken.

log("AuthRepository.get_user_by_username", "EXPLORE",
    "User not found, returning None", {"username": username}, error="User does not exist in database")

Quick Reference

Situation	Marker	Level	error field
About to execute DB query	REASON	INFO	—
DB query returned results	REFLECT	INFO	—
DB query returned empty set (happy path)	REFLECT	INFO	—
DB query failed, fallback to cache	EXPLORE	WARNING	required
About to call external API	REASON	INFO	—
API responded 200	REFLECT	INFO	—
API responded 5xx, retrying	EXPLORE	WARNING	required
API exhausted retries	EXPLORE	ERROR	required
Precondition check fails (e.g., not found)	EXPLORE	WARNING	required
State validation passes	REFLECT	INFO	—
Decomposing a complex loop iteration	REASON	DEBUG	—

Never use generic tags like Entry, Exit, Action, Coherence:OK/FAIL. Those are replaced entirely by the molecular bond markers.

III. Trace Propagation (Python Implementation)

import uuid
import logging
from contextvars import ContextVar
from datetime import datetime, timezone

# ── Trace context ────────────────────────────────────────────
_trace_id: ContextVar[str] = ContextVar("trace_id", default="")
_span_id: ContextVar[str] = ContextVar("span_id", default="")

def seed_trace_id() -> str:
    """Call once at request/job entry to initialise the trace."""
    tid = uuid.uuid4().hex
    _trace_id.set(tid)
    _span_id.set("")  # reset span
    return tid

def get_trace_id() -> str:
    return _trace_id.get()

def push_span(span: str) -> str:
    """Set a new span_id (e.g. function name). Returns the previous span for restore."""
    prev = _span_id.get()
    _span_id.set(span)
    return prev

def pop_span(prev: str) -> None:
    _span_id.set(prev)

# ── Structured logger ────────────────────────────────────────
_logger = logging.getLogger("cot")

def log(
    src: str,
    marker: str,
    intent: str,
    payload: dict | None = None,
    error: str | None = None,
    level: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
) -> None:
    """Emit a single molecular CoT log line.

    Args:
        src:      Qualified function name, e.g. "AuthRepository.get_user"
        marker:   One of "REASON", "REFLECT", "EXPLORE"
        intent:   One-line description of the step's purpose
        payload:  Arbitrary key-value data (params, result snippet)
        error:    Required for EXPLORE; describes the violated assumption
        level:    Override log level (inferred from marker if omitted)
        trace_id: Override trace_id (auto-picked from ContextVar if omitted)
        span_id:  Override span_id (auto-picked from ContextVar if omitted)
    """
    # Infer level from marker if not overridden
    if level is None:
        if marker == "EXPLORE":
            level = "WARNING"
        else:
            level = "INFO"

    record = {
        "ts":       datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "level":    level,
        "trace_id": trace_id or _trace_id.get(),
        "src":      src,
        "marker":   marker,
        "intent":   intent,
    }

    if span_id or (sid := _span_id.get()):
        record["span_id"] = span_id or sid
    if payload is not None:
        record["payload"] = payload
    if error is not None:
        record["error"] = error

    # Map level string to logging constant
    _logger.log(
        getattr(logging, level.upper(), logging.INFO),
        "%s", json.dumps(record, ensure_ascii=False, default=str),
    )

FastAPI middleware (trace seeding)

from starlette.middleware.base import BaseHTTPMiddleware

class TraceMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        seed_trace_id()
        response = await call_next(request)
        return response

IV. Python Decorator (Span + Marker)

For C4/C5 functions, a decorator that auto-emits REASON / REFLECT markers:

import asyncio
from functools import wraps

def cot_span(marker: str = "REASON", intent: str | None = None):
    """Wrap a function in a CoT span.  On enter → REASON, on success → REFLECT,
    on exception → EXPLORE."""
    def decorator(func):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            src = f"{func.__module__}.{func.__qualname__}"
            prev_span = push_span(func.__qualname__)
            default_intent = intent or f"Execute {func.__qualname__}"
            try:
                log(src, marker, default_intent, payload=_summarise_args(args, kwargs))
                result = await func(*args, **kwargs)
                log(src, "REFLECT", f"{func.__qualname__} completed",
                    payload={"result": _summarise_value(result)})
                return result
            except Exception as e:
                log(src, "EXPLORE", f"{func.__qualname__} failed",
                    error=str(e), payload={"args": _summarise_args(args, kwargs)})
                raise
            finally:
                pop_span(prev_span)

        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            ...  # same logic, sync variant
        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    return decorator


def _summarise_value(val, max_len: int = 200) -> str:
    s = str(val)
    return s[:max_len] + "..." if len(s) > max_len else s

def _summarise_args(args, kwargs) -> dict:
    # Skip 'self', 'cls', 'db', 'request' — too verbose
    skip = {"self", "cls", "db", "request", "session"}
    result = {}
    for k, v in kwargs.items():
        if k not in skip:
            result[k] = _summarise_value(v)
    return result

V. Svelte / Frontend Pattern

The frontend implementation lives at frontend/src/lib/cot-logger.ts (installed as $lib/cot-logger).

API

function log(
  src: string,          // e.g. "MigrationModel.executeMigration"
  marker: LogMarker,    // "REASON" | "REFLECT" | "EXPLORE"
  intent: string,       // human-readable one-liner
  payload?: Record<string, unknown>,  // params, result snippet
  error?: string,       // required for EXPLORE
): void;

Import

import { log, setTraceId, getTraceId } from "$lib/cot-logger";

Usage in a Svelte component

<script lang="ts">
import { log } from "$lib/cot-logger";
import { fetchApi } from "$lib/api";

let { jobId }: { jobId: string } = $props();

async function loadJob(): Promise<void> {
    log("JobDetail", "REASON", "Fetch job details", { jobId });

    try {
        const resp = await fetchApi(`/api/jobs/${jobId}`);
        if (!resp.ok) throw new Error(`Status ${resp.status}`);

        const data = await resp.json();
        log("JobDetail", "REFLECT", "Job details loaded",
            { rows: data.records?.length });
        return data;
    } catch (e: unknown) {
        log("JobDetail", "EXPLORE", "Failed to load job",
            { jobId }, e instanceof Error ? e.message : "Unknown");
        throw e;
    }
}
</script>

trace_id Propagation

The trace ID is set automatically when the backend returns it. Call setTraceId(id) manually if needed:

import { setTraceId } from "$lib/cot-logger";
import { requestApi } from "$lib/api";

const res = await requestApi("/api/endpoint");
if (res.trace_id) setTraceId(res.trace_id);

VI. CLI / Stdout Reader (for humans)

To make JSON lines readable in development:

# Pretty-print the last 50 CoT lines
tail -50 app.log | python3 -c "
import sys, json
for line in sys.stdin:
    line = line.strip()
    if not line: continue
    rec = json.loads(line)
    m = rec.get('marker','?')
    icon = {'REASON':'→','REFLECT':'✓','EXPLORE':'⚠'}.get(m, '·')
    err = f\" | {rec['error']}\" if 'error' in rec else ''
    pay = f\" | {rec.get('payload','')}\" if 'payload' in rec else ''
    print(f\"{icon} {rec['level']:7} {rec['src']}  — {rec['intent']}{pay}{err}\")
"

VII. Anti-patterns

❌ Don't	✅ Do
COHERENCE:OK on happy path	REFLECT with verification summary
Action: something	REASON with intent
Entry / Exit	REASON at entry, REFLECT at exit
Wrapping EVERY line with a marker	Only log semantically meaningful steps
Plain-text log lines	Always JSON lines
marker without intent	Every marker has a human-readable intent
Logging raw passwords or tokens in payload	Always sanitise sensitive data
Spread markers across multiple modules without trace_id	Always propagate trace_id

#endregion MolecularCoTLogging