Files
wehub-resource-sync 4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

130 lines
4.2 KiB
Python

"""Formatting helpers for streamed investigation output."""
from __future__ import annotations
import math
import re
from collections.abc import Sequence
from typing import Any
from config.constants.investigation import MAX_INVESTIGATION_LOOPS
def format_prior_tools_clause(
tools: Sequence[str],
*,
max_tools: int = 3,
) -> str:
"""Appendix naming tools gathered since the previous LLM lap."""
if not tools:
return ""
labels: list[str] = []
counts: dict[str, int] = {}
for label in tools:
stripped = label.strip()
if not stripped:
continue
counts[stripped] = counts.get(stripped, 0) + 1
if stripped not in labels:
labels.append(stripped)
if not labels:
return ""
rendered = [
f"{label} x{counts[label]}" if counts[label] > 1 else label for label in labels[:max_tools]
]
suffix = ", ..." if len(labels) > max_tools else ""
return f" after {', '.join(rendered)}{suffix}"
def investigation_llm_progress_hint(
iteration: int,
*,
max_loops: int = MAX_INVESTIGATION_LOOPS,
prior_tools: Sequence[str] | None = None,
) -> str:
"""Human-readable status for one investigation-agent LLM lap.
Each ``llm_start`` event maps to one ReAct think step: the model reads
accumulated alert + tool evidence and either requests more tools or stops.
"""
lap = iteration + 1
cap = f"lap {lap}/{max_loops}"
tools_clause = format_prior_tools_clause(prior_tools or ())
if iteration == 0:
return f"Planning investigation ({cap}){tools_clause}"
return f"Reviewing evidence ({cap}){tools_clause}"
def _clean_markdown_line(line: str) -> str:
"""Strip both bulleted lists (•, ●, -, —, *) and numbered lists (e.g. 1., 2))."""
stripped = line.strip()
prev = ""
while stripped != prev:
prev = stripped
stripped = re.sub(r"^[-•●—]\s+", "", stripped)
# Markdown ``* item`` list marker only — not ``*Italic Section:*`` headings.
stripped = re.sub(r"^\*\s+", "", stripped)
stripped = re.sub(r"^\d+[.)]\s+", "", stripped)
return stripped
def _normalized_report_heading_inner(line: str) -> str:
"""Normalize LLM report lines for heading keyword matching."""
s = line.strip()
while s.startswith("#"):
s = s[1:].strip()
if s.startswith("**"):
core = s[2:]
if core.endswith("**:"):
core = core[:-3]
elif core.endswith("**"):
core = core[:-2]
return core.strip()
if len(s) >= 2 and s.startswith("[") and s.endswith("]") and ":" not in s:
return s[1:-1].strip()
if (
len(s) >= 3
and s.startswith("*")
and s.endswith("*")
and not s.startswith("* ")
and "**" not in s
):
inner = s[1:-1].strip()
if ":" in inner or len(inner.split()) >= 3:
return inner
return s.strip()
def _report_line_looks_like_heading(line: str, *, inner: str) -> bool:
"""True if the line uses a heading-like structure (not prose)."""
stripped = line.strip()
if stripped.startswith("#"):
return True
is_bracket = (
stripped.startswith("[") and stripped.rstrip().endswith("]") and ":" not in stripped
)
is_bold_md = stripped.startswith("**") and (stripped.endswith("**") or stripped.endswith("**:"))
wrapped_ast = (
len(stripped) >= 3
and stripped.startswith("*")
and stripped.endswith("*")
and not stripped.startswith("* ")
and "**" not in stripped
and (":" in stripped[1:-1] or len(stripped[1:-1].strip().split()) >= 3)
)
shouty = inner.isupper() and len(inner.replace(" ", "")) >= 8 and len(inner.split()) <= 14
return bool(is_bracket or is_bold_md or wrapped_ast or shouty)
def _validity_score_percent(score: Any) -> str | None:
"""Format a 0..1 validity score for display, or None if the payload is unusable."""
if score is None or isinstance(score, bool):
return None
if not isinstance(score, (int, float)):
return None
v = float(score)
if not math.isfinite(v):
return None
v = max(0.0, min(1.0, v))
return f"{int(v * 100)}%"