파일 하나가 아니라 세그먼트

def seal_active_segment(sessions_dir: Path, conversation_id: str, *, threshold: int | None = None) -> Path | None:
    """Seal the active file into the next segment when it has reached the
    threshold. Returns the sealed path, or None when nothing was sealed.

    The caller guarantees no writer holds the conversation (SessionLogger
    calls this under its process lock, with no logger open for it). The
    rename is atomic on one filesystem; the next writer creates a fresh
    active file.
    """
    active = conv_jsonl_path(sessions_dir, conversation_id)
    try:
        size = active.stat().st_size
    except FileNotFoundError:
        return None
    if size < (segment_bytes() if threshold is None else threshold):
        return None
    directory = segments_dir(sessions_dir, conversation_id)
    directory.mkdir(parents=True, exist_ok=True)
    existing = sealed_segment_paths(sessions_dir, conversation_id)
    last = int(_SEGMENT_NAME_RE.match(existing[-1].name).group(1)) if existing else 0
    target = directory / f"{last + 1:06d}.jsonl"
    if target.exists():
        raise FileExistsError(target)
    os.rename(active, target)
    logger.info("Sealed JSONL segment %s (%d bytes) for %s", target.name, size, conversation_id)
    return target

# A minimal sketch of a segmented, append-only conversation log.
import json
import os
import tempfile
from pathlib import Path

SEGMENT_BYTES = 200  # tiny so the demo seals; the engine's default is 8 MiB


def active_path(root: Path, cid: str) -> Path:
    return root / f"{cid}.jsonl"


def sealed_paths(root: Path, cid: str) -> list[Path]:
    directory = root / f"{cid}.segments"
    return sorted(directory.glob("*.jsonl")) if directory.is_dir() else []


def open_logger(root: Path, cid: str):
    """Seal only when a logger opens, never mid-turn: one logger writes one whole turn."""
    active = active_path(root, cid)
    if active.exists() and active.stat().st_size >= SEGMENT_BYTES:
        directory = root / f"{cid}.segments"
        directory.mkdir(exist_ok=True)
        target = directory / f"{len(sealed_paths(root, cid)) + 1:06d}.jsonl"
        os.rename(active, target)  # atomic on one filesystem; the file moves whole
    return open(active, "a", encoding="utf-8")


def write_turn(root: Path, cid: str, turn: int) -> None:
    with open_logger(root, cid) as log:
        for kind in ("user", "text_delta", "text_delta", "done"):
            log.write(json.dumps({"turn": turn, "type": kind}) + "\n")
            log.flush()  # on disk before it would be shown


def conversation_paths(root: Path, cid: str) -> list[Path]:
    """Oldest first: the sealed segments, then the active file."""
    active = active_path(root, cid)
    return sealed_paths(root, cid) + ([active] if active.exists() else [])


with tempfile.TemporaryDirectory() as tmp:
    root = Path(tmp)
    for turn in range(1, 8):
        write_turn(root, "chief", turn)

    naive = sum(len(p.read_text().splitlines()) for p in root.glob("*.jsonl"))
    events = [json.loads(line) for p in conversation_paths(root, "chief") for line in p.read_text().splitlines()]
    print(f"a glob of the directory sees {naive} lines")
    print(f"the segment-aware reader sees {len(events)} lines, turns {sorted({e['turn'] for e in events})}")
    for p in conversation_paths(root, "chief"):
        turns = sorted({json.loads(line)["turn"] for line in p.read_text().splitlines()})
        print(f"  {p.relative_to(root)}: {p.stat().st_size} bytes, turns {turns}")