script-build-and-framework-tmp/visualization/backend/app/module_audit/log_context.py

from __future__ import annotations

import re
from typing import Any


_ANCHOR_RE = re.compile(
    r"^\[(?:MSG|TOOL)_ANCHOR:msg_id=([^:\]]+):seq=(\d+)(?::[^\]]+)?\]\s*$",
    re.MULTILINE,
)
_BLOCK_START_RE = re.compile(r"^\[(?:AGENT_TASK|FOLD):[^\]]+\]\s*$", re.MULTILINE)


def locate_log_module(log_content: str | None, event: dict[str, Any]) -> dict[str, Any] | None:
    """Return only the anchored log module associated with an event."""

    if not log_content:
        return None
    msg_id = str(event.get("msg_id") or "").strip()
    if not msg_id:
        return None
    anchors = list(_ANCHOR_RE.finditer(log_content))
    matched = next(
        (
            match
            for match in anchors
            if match.group(1) == msg_id
        ),
        None,
    )
    if matched is None:
        return None

    previous_starts = [
        item
        for item in _BLOCK_START_RE.finditer(log_content, max(0, matched.start() - 2_000), matched.start())
    ]
    start = previous_starts[-1].start() if previous_starts else matched.start()
    next_anchor = next((item for item in anchors if item.start() > matched.start()), None)
    end = next_anchor.start() if next_anchor else len(log_content)
    content = log_content[start:end].strip()
    return {
        "anchor": matched.group(0),
        "msgId": matched.group(1),
        "sequence": int(matched.group(2)),
        "content": content,
        "characters": len(content),
    }


def _integer(value: Any) -> int | None:
    try:
        return int(value) if value is not None else None
    except (TypeError, ValueError):
        return None