refactor(log_registry): Session dataclass already in place; verified no dict-style consumers

This commit is contained in:
ed
2026-06-24 17:19:28 -04:00
parent 25a2205722
commit 6956676f7c
9 changed files with 149 additions and 90 deletions
@@ -0,0 +1,4 @@
from src.mcp_client import get_tool_schemas
schemas = get_tool_schemas()
print(f"get_tool_schemas returned {len(schemas)} entries")
print(f"First: {schemas[0]['name']}")
@@ -0,0 +1,11 @@
from src.provider_state import get_history
h = get_history("anthropic")
h.append({"role": "user", "content": "hi"})
h.append({"role": "assistant", "content": "hello"})
print(f"len: {len(h)}")
print(f"bool: {bool(h)}")
roles = [m["role"] for m in h]
print(f"iter: {roles}")
print(f"getitem: {h[0]}")
h.clear()
print(f"after clear len: {len(h)}")
@@ -0,0 +1,28 @@
import sys
sys.path.insert(0, ".")
import ast
from pathlib import Path
# Strict: find functions where a parameter is DIRECTLY typed as Metadata (not nested)
for fpath in Path("src").glob("*.py"):
src = fpath.read_text(encoding="utf-8")
tree = ast.parse(src)
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for arg in node.args.args + node.args.kwonlyargs:
if arg.annotation is None:
continue
ann_str = ast.unparse(arg.annotation)
is_metadata_direct = ann_str in ("Metadata", "dict[str, Any]", "Optional[Metadata]", "Optional[dict[str, Any]]")
if not is_metadata_direct:
continue
# Check if there's a nil-check on this parameter
for sub in ast.walk(node):
if isinstance(sub, ast.Compare):
left = sub.left
if isinstance(left, ast.Name) and left.id == arg.arg:
for c in sub.comparators:
if isinstance(c, ast.Constant) and c.value is None:
print(f" {fpath.name}:{node.lineno} {node.name} - param={arg.arg} ann={ann_str} nil@{sub.lineno}")
break
@@ -0,0 +1,13 @@
import sys
sys.path.insert(0, ".")
from src.code_path_audit_ssdl import detect_nil_check_pattern
from src.code_path_audit import build_pcg
r = build_pcg("src")
pcg = r.data
metadata_consumers = pcg.consumers.get("Metadata", [])
nil_funcs = [f for f in metadata_consumers if detect_nil_check_pattern(f, "src")]
print(f"Total Metadata consumers with nil-checks: {len(nil_funcs)}")
for f in nil_funcs:
print(f" - {f.fqname} @ {f.file}:{f.line}")
@@ -0,0 +1,30 @@
import sys
sys.path.insert(0, ".")
import ast
from pathlib import Path
for fpath in ("src/aggregate.py", "src/ai_client.py"):
p = Path(fpath)
src = p.read_text(encoding="utf-8")
tree = ast.parse(src)
print(f"=== {fpath} ===")
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
has_nil = False
nil_vars = []
for sub in ast.walk(node):
if isinstance(sub, ast.Compare):
for ci, c in enumerate(sub.comparators):
if isinstance(c, ast.Constant) and c.value is None:
has_nil = True
left = sub.left
if isinstance(left, ast.Name):
nil_vars.append((left.id, sub.lineno))
else:
nil_vars.append(("?", sub.lineno))
if has_nil:
# Check parameters
params = []
for arg in node.args.args + node.args.kwonlyargs:
params.append(arg.arg)
print(f" line {node.lineno}: {node.name} - nil_vars: {nil_vars[:5]}, params: {params[:8]}")
@@ -0,0 +1,14 @@
import sys
sys.path.insert(0, ".")
from src.code_path_audit_ssdl import detect_nil_check_pattern
from src.code_path_audit import FunctionRef
fref = FunctionRef(
fqname="src.aggregate._build_files_section_from_items",
file="aggregate.py",
line=300,
role="consumer",
)
result = detect_nil_check_pattern(fref, "src")
print(f"detect_nil_check_pattern(_build_files_section_from_items) = {result}")
print("PASS" if not result else "FAIL")
@@ -0,0 +1,49 @@
import sys
sys.path.insert(0, ".")
from src.code_path_audit_ssdl import compute_effective_codepaths
from src.code_path_audit import build_pcg, FunctionRef
from src.code_path_audit_analysis import aggregate_pattern_from_consumers
from src.code_path_audit_cross_audit import (
aggregate_findings,
build_cross_audit_findings_for_aggregate,
)
from src.code_path_audit_analysis import (
compute_real_type_alias_coverage,
compute_real_decomposition_cost,
extract_real_optimization_candidates,
)
from src.code_path_audit import AggregateProfile, ResultCoverage, TypeAliasCoverage, CrossAuditFindings, DecompositionCost, FrequencyEvidence
from src.code_path_audit import classify_memory_dim
pcg_result = build_pcg("src")
pcg = pcg_result.data
producers = tuple(pcg.producers.get("Metadata", []))
consumers = tuple(pcg.consumers.get("Metadata", []))
print(f"Producers: {len(producers)}")
print(f"Consumers: {len(consumers)}")
profile = AggregateProfile(
name="Metadata",
aggregate_kind="typealias",
memory_dim=classify_memory_dim("Metadata", producers[0].file if producers else "", {}),
producers=producers,
consumers=consumers,
access_pattern="mixed",
access_pattern_evidence=(),
frequency="per_turn",
frequency_evidence=(),
result_coverage=ResultCoverage(0, 0, 0, 0, ""),
type_alias_coverage=TypeAliasCoverage(0, 0, 0, ""),
cross_audit_findings=CrossAuditFindings((), (), (), (), ()),
decomposition_cost=DecompositionCost(0, 0, 0, "insufficient_data", "", None, 0, False),
optimization_candidates=(),
is_candidate=False,
)
ec = compute_effective_codepaths(profile, "src")
print(f"Effective codepaths: {ec}")
print(f"Baseline: 4.01e22")
print(f"Drop: {4.01e22 - ec}")
print(f"Drop %: {(4.01e22 - ec) / 4.01e22 * 100:.6f}%")
print(f"VC4: {'PASS' if ec <= 4.01e22 * 0.9 else 'FAIL'} (need 10% drop)")