import re

nums = [int([Link](1)) for f in sub_batch_files

if (m := [Link](r"sub_batch_(\d+)\.jsonl$", f))]

max_index = max(nums) # or max(nums, default=1)

[Link](r"sub_batch_(\d+)\.jsonl$", f) tries to match this pattern in each

filename f:

sub_batch_ — literal text.

(\d+) — one or more digits, captured as group 1. (\d = digit, + = one or

more.)

\.jsonl — literal .jsonl (the dot is escaped so it means a real dot, not “any
char”).

$ — end of the string anchor (ensures the filename ends with .jsonl and
nothing extra after).

# blocking_demo.py

import asyncio

import time

def slow_function(x: int) -> int:

# Simulate heavy/blocking work (CPU or a blocking I/O/lib call)

[Link](2) # Blocks the OS thread

return x * 2
async def ticker():

# Shows whether the event loop is alive

i=0

while True:

print(f"[ticker] alive {i}")

i += 1

await [Link](0.2)

async def main():

# Start a background ticker

tick_task = asyncio.create_task(ticker())

print("[main] starting 5 blocking calls (this will FREEZE the ticker)")

results = []

for i in range(5):

# ❌ This is synchronous and blocks the event loop thread

r = slow_function(i)

[Link](r)

print(f"[main] blocking result for {i} -> {r}")

tick_task.cancel()

try:

await tick_task

except [Link]:

pass

print("[main] done, results:", results)

if __name__ == "__main__":

[Link](main())

What you’ll see

The ticker prints a couple of lines like [ticker] alive 0, [ticker] alive 1 … then
suddenly stops for ~10 seconds (5 calls × 2s each).

After all blocking work finishes, the ticker may resume briefly or immediately
cancel when main ends.

Non-blocking version (event loop stays responsive)

# nonblocking_demo.py

import asyncio

import time

from typing import List

def slow_function(x: int) -> int:

[Link](2) # Same blocking call as before

return x * 2

async def ticker():

i=0

while True:

print(f"[ticker] alive {i}")

i += 1
 await [Link](0.2)

async def compute_many(nums: List[int]):

# Kick off the slow work in threads WITHOUT blocking the event loop

# You can run them sequentially with to_thread...

# results = []

# for n in nums:

# [Link](await asyncio.to_thread(slow_function, n))

# return results

# ...or concurrently with gather:

tasks = [asyncio.to_thread(slow_function, n) for n in nums]

return await [Link](*tasks)

async def main():

tick_task = asyncio.create_task(ticker())

print("[main] starting 10 non-blocking calls via asyncio.to_thread")

results = await compute_many(list(range(10)))

print("[main] done, results:", results)

tick_task.cancel()

try:

await tick_task

except [Link]:

pass
if __name__ == "__main__":

[Link](main())

What you’ll see

The ticker never stops; you keep seeing [ticker] alive N every ~0.2s while the
work runs.

Because we used asyncio.to_thread, the blocking [Link](2) happens in

worker threads, so the event loop can keep scheduling other coroutines (like
ticker).

Bonus: Throttled concurrency (limit threads)

If you want many tasks but don’t want to spawn too many threads at once:

# throttled_nonblocking_demo.py

import asyncio

import time

from typing import Iterable

def slow_function(x: int) -> int:

[Link](2)

return x * 2

async def ticker():

i=0

while True:

print(f"[ticker] alive {i}")

 i += 1

await [Link](0.2)

async def bounded_to_thread(fn, arg, sem: [Link]):

async with sem:

return await asyncio.to_thread(fn, arg)

async def main():

tick_task = asyncio.create_task(ticker())

nums = list(range(30))

sem = [Link](5) # at most 5 blocking jobs running

concurrently

tasks = [bounded_to_thread(slow_function, n, sem) for n in nums]

results = await [Link](*tasks)

print("[main] done, results (len):", len(results))

tick_task.cancel()

try:

await tick_task

except [Link]:

pass

if __name__ == "__main__":

[Link](main())
Add a special case for [Link] and convert it via .tolist() before
normalizing each item:

if isinstance(raw, [Link]):

return [str(item).strip() for item in [Link]() if str(item).strip()]

Here’s the gist on the double-underscore (“dunder”) names:

 What “dunder” means

 Names wrapped in double underscores (like __init__) are

“special” to Python. The interpreter looks for these exact names
to implement built-in behaviors (operators, construction,
iteration, context managers, etc.).

 __init__ vs __init__.py

 __init__(self, …) is a special method on classes. Python calls it

right after creating an instance (the initializer; the actual
constructor is __new__).

 __init__.py is a package initializer file. When you import a

package, Python executes this file to set up the package
namespace (commonly where packages set __version__). With
PEP 420, it’s no longer required for “namespace packages”, but
most normal packages still include it.

 __version__

 Not enforced by the interpreter; it’s a widely adopted convention

for packages/modules to expose their version string.

 Where it lives: often set in the top-level package __init__.py,

e.g., __version__ = "1.2.3".

 Canonical way to get installed distribution versions

is [Link]("package-name"), which reads
package metadata without importing the package.

 Why double underscores?

 It signals “this name has special semantics” and reduces

collisions with ordinary attribute names. The interpreter and
 standard tooling check for these exact names
(e.g., __len__, __iter__, __enter__/__exit__, __add__, __repr__, __get
attr__, etc.).

 Don’t confuse with name mangling

 A name with only a double leading underscore (e.g., __private)

inside a class triggers name-mangling to _ClassName__private.
That’s different from special dunders which have both leading
and trailing underscores (e.g., __init__, __version__) and are not
mangled.

 Markdown “bolding”

 Some UIs render __name__ as bold because Markdown treats

double underscores as bold markers. In a Python shell it won’t
bold; it’s just a display artifact.

 Practical tips

 Show versions safely:

 python -c "import [Link] as m;

print([Link]('lancedb'), [Link]('pyarrow'))"

 or python -c "import lancedb, pyarrow;

print(lancedb.__version__, pyarrow.__version__)" (imports
the packages, which can have side effects)

 Common module dunders you’ll

see: __name__, __file__, __package__, __doc__, __all__, __spec__.

Using pandas at my finalize_graph.py so it prepares node

metadata and flattened attributes, DataFrame, flatten
arrays/lists/dicts

# Copyright (c) 2024 Microsoft Corporation.

# Licensed under the MIT License

"""A module containing run_workflow method definition."""

import logging

from typing import Any

import pandas as pd

from [Link].embed_graph_config import EmbedGraphConfig

from [Link].graph_rag_config import GraphRagConfig

from [Link].create_graph import create_graph

from [Link].finalize_entities import finalize_entities

from [Link].finalize_relationships import finalize_relationships

from [Link].snapshot_graphml import snapshot_graphml

from [Link] import PipelineRunContext

from [Link] import WorkflowFunctionOutput

from [Link] import load_table_from_storage, write_table_to_storage

logger = [Link](__name__)

def _flatten_value(value: Any) -> Any:

"""Convert complex values (lists, dicts, ndarrays) to GraphML-friendly

scalars/strings."""

try:

import numpy as np # local import to avoid hard dependency at module

import time

except Exception:

np = None

# Pass through simple scalars

if value is None or isinstance(value, (str, int, float, bool)):

return value
# numpy types

if np is not None:

if isinstance(value, ([Link],)):

return [Link]()

if isinstance(value, ([Link],)):

try:

# Flatten small arrays sensibly

return ", ".join(str(x) for x in [Link]().tolist())

except Exception:

return str(value)

# lists/tuples

if isinstance(value, (list, tuple)):

try:

return ", ".join(str(v) for v in value)

except Exception:

return str(value)

# dicts

if isinstance(value, dict):

try:

# Compact JSON-like representation

return ", ".join(f"{k}={v}" for k, v in [Link]())

except Exception:

return str(value)

# Fallback
 return str(value)

def _prepare_entities_for_graphml(entities: [Link]) -> [Link]:

"""Return a copy of entities with flattened, GraphML-friendly columns.

- Ensures a 'text' column (alias of 'description') for downstream matching

- Flattens list/array/dict-like columns to strings

- Keeps core descriptive fields readable

"""

if entities is None or len(entities) == 0:

return entities

df = [Link]()

# Provide a 'text' alias to help downstream consumers that expect [Link]

if "text" not in [Link]:

if "description" in [Link]:

df["text"] = df["description"].fillna("")

else:

df["text"] = ""

# Flatten all columns to GraphML-friendly scalars

for col in [Link]:

df[col] = df[col].apply(_flatten_value)

return df

async def run_workflow(

 config: GraphRagConfig,

context: PipelineRunContext,

) -> WorkflowFunctionOutput:

"""All the steps to create the base entity graph."""

[Link]("Workflow started: finalize_graph")

entities = await load_table_from_storage("entities", context.output_storage)

relationships = await load_table_from_storage(

"relationships", context.output_storage

final_entities, final_relationships = finalize_graph(

entities,

relationships,

embed_config=config.embed_graph,

layout_enabled=[Link],

await write_table_to_storage(final_entities, "entities", context.output_storage)

await write_table_to_storage(

final_relationships, "relationships", context.output_storage

if [Link]:

# Build a graph that includes node attributes and useful edge metadata.

nodes_df = _prepare_entities_for_graphml(final_entities)

# Keep common edge attrs (weight + description if present)

edge_attrs: list[str] = ["weight"]

 if "description" in final_relationships.columns:

edge_attrs.append("description")

graph = create_graph(

final_relationships,

edge_attr=edge_attrs,

nodes=nodes_df,

node_id="title",

# Log a concise snapshot summary for easier debugging downstream

try:

[Link](

"Snapshotting GraphML: nodes=%d, edges=%d, edge_attrs=%s",

len(nodes_df) if nodes_df is not None else 0,

len(final_relationships),

",".join(edge_attrs),

except Exception:

pass

await snapshot_graphml(

graph,

name="graph",

storage=context.output_storage,

[Link]("Workflow completed: finalize_graph")

 return WorkflowFunctionOutput(

result={

"entities": entities,

"relationships": relationships,

def finalize_graph(

entities: [Link],

relationships: [Link],

embed_config: EmbedGraphConfig | None = None,

layout_enabled: bool = False,

) -> tuple[[Link], [Link]]:

"""All the steps to finalize the entity and relationship formats."""

final_entities = finalize_entities(

entities, relationships, embed_config, layout_enabled

final_relationships = finalize_relationships(relationships)

return (final_entities, final_relationships)

Project IMPLEMENTING TRACEABILITY MENTIS

+from bisect import bisect_right

 O(log L) line lookup. With bisect_right(self._line_starts, pos), you map any char
offset to a 1-based line number quickly. This is a canonical trick in text editors and
compilers.

implementing traceability

 No lossy tokenization. Using \S+ leaves whitespace intact in the original

buffer, which is crucial for slicing true spans later.

b) Chunk construction: _create_chunks(...)

 For spec text, chunks are created by slicing the original text using
word_spans[i:j] bounds—not by ' '.join(words)—so each chunk is an exact
substring of the persisted spec_text.txt.

 For each chunk i, it records:

o self.chunk_spans_chars[i] = (start_char, end_char)

o self.chunk_spans_lines[i] = (start_line, end_line)

 It also includes a fallback (no spans) when word spans aren’t available.

  Referential integrity. By slicing text[start:end], you guarantee

text[start:end] == chunks[i]. This is the single most important invariant for
provenance.
  Overlap safety. The step uses max(1, chunk_size - overlap) to avoid
zero/negative stepping (a classic off-by-one pitfall).

Retrieval: retrieve(query, k) returns (chunk_index, chunk_text)

 A TF-IDF vectorizer is fit on [Link]. Given a query, it computes cosine

similarities and returns the top-k results as pairs (i, [Link][i]).

 Returning the index is key: it ties straight back to chunk_spans_chars[i] and

chunk_spans_lines[i].

+ results = [Link](query, k=k)

+ for i, (chunk_idx, chunk) in enumerate(results):

# Score based on rank and chunk properties

@@ -145,2 +145,23 @@

+ # Build metadata including provenance

+ meta = {

+ 'chunk_size': retriever.chunk_size,
 + 'overlap': [Link],

+ 'rank': i + 1,

+ 'content_type': 'spec',

+ 'chunk_index': chunk_idx,

+ }

+ if (

+ hasattr(retriever, 'chunk_spans_chars') and

+ retriever.chunk_spans_chars and 0 <= chunk_idx <

len(retriever.chunk_spans_chars)

+ ):

+ start_char, end_char =
retriever.chunk_spans_chars[chunk_idx]

+ meta['span_chars'] = [start_char, end_char]

+ if (

+ hasattr(retriever, 'chunk_spans_lines') and

+ retriever.chunk_spans_lines and 0 <= chunk_idx <

len(retriever.chunk_spans_lines)

+ ):

+ start_line, end_line =
retriever.chunk_spans_lines[chunk_idx]

+ meta['span_lines'] = [start_line, end_line]

🧠 Background — why JSON serialization fails

When you use:

top_k_indices = [Link]()[-k:][::-1]

argsort() from NumPy returns a [Link] containing NumPy scalar types,

e.g.:

>>> type(top_k_indices[0])
<class 'numpy.int64'>

The problem is that numpy.int64 is not JSON-serializable by Python’s built-in

json module.

So if downstream code tries something like:

[Link]({"chunk_index": top_k_indices[0]}, f)

you get:

TypeError: Object of type int64 is not JSON serializable

That’s what this patch is fixing.

🧩 The change explained

Original

return [(i, [Link][i]) for i in top_k_indices]

 i here is a NumPy integer (e.g. numpy.int64).

 Works fine for arithmetic.

 Fails later if this tuple is ever passed into something like [Link](...), or
stored inside [Link] that will be serialized.

Fixed version

# Return (index, chunk_text) for provenance with Python int indices

# Ensure we don't leak numpy.int64 into downstream JSON

try:

idx_list = top_k_indices.tolist()

except Exception:

idx_list = list(top_k_indices)

return [(int(i), [Link][int(i)]) for i in idx_list]

Step by step:

1. top_k_indices.tolist()

o Converts the NumPy array into a regular Python list of int values.

o Example:
 o [Link]([1,2,3], dtype=np.int64).tolist() # → [1, 2, 3]

o If for any reason top_k_indices is not a NumPy array (already a list or

something else), .tolist() might fail → hence the try/except.

2. list(top_k_indices) (fallback)

o Works for most iterable types.

3. int(i)

o Explicitly casts every index into a native Python int.

o Guarantees that downstream code, including JSON serialization,

handles these values safely.

4. Final return:

5. return [(int(i), [Link][int(i)]) for i in idx_list]

o Each element is now a tuple (int, str).

o Both types are natively serializable by [Link]().

🔍 Example of what it prevents

Before:

top_k_indices = [Link]([42, 43])

results = [(top_k_indices[0], "some text")]

[Link](results)

→ ❌ TypeError: Object of type int64 is not JSON serializable

After:

results = [(int(top_k_indices[0]), "some text")]

[Link](results)

→ ✅ Works perfectly.

⚙️Python principle behind this

🔹 Numpy types vs built-in types

NumPy defines its own scalar classes (numpy.int64, numpy.float32, etc.),

which mimic Python’s built-in int and float but have C-level dtypes and are not
part of the standard JSON encoder.
🔹 JSON encoding in Python

[Link]() (and [Link]()) use the default [Link], which supports:

 built-in Python types (int, float, str, list, dict, bool, None)

 not 3rd-party numeric types.

Hence, always cast to native types before serialization.

✅ Why the fix is good practice

Problem Fix Benefit

NumPy scalar not JSON Convert via .tolist() or

Ensures stable JSON
serializable int()

Downstream context
Explicit Python int Prevents pipeline crash
metadata fails

Portability issues (different Works across OS/Python

Explicit conversion
dtypes) versions

# src/indexing/protocol_index.py

import re

from dataclasses import dataclass

from typing import List, Dict

@dataclass

class ProtocolPara:

id: str

text: str

tokens_hex: List[str]

tokens_words: List[str]

timing_mentions: List[str]
 numbers: List[int] # plain digits found

class ProtocolIndex:

HEX_RE = [Link](r"0x[0-9A-Fa-f]+")

DIGITS_RE = [Link](r"\b\d+\b")

TIMING_WORDS = {"cycle", "clock", "latency", "period", "baud", "MHz",

"kHz"}

SPECIAL_WORDS = {"ACK", "CR", "LF"}

def __init__(self, paragraphs: List[str], prefix: str = "txt:protocol"):

[Link]: List[ProtocolPara] = []

for i, txt in enumerate(paragraphs, start=1):

pid = f"{prefix}:para{i}"

hexes = self.HEX_RE.findall(txt)

nums = list(map(int, self.DIGITS_RE.findall(txt)))

timing = [w for w in self.TIMING_WORDS if [Link](rf"\b{w}\b", txt,

re.I)]

specials = [w for w in self.SPECIAL_WORDS if [Link](rf"\b{w}\b",

txt, re.I)]

[Link](ProtocolPara(pid, txt, hexes, specials, timing,

nums))

def query(self, terms: List[str]) -> List[ProtocolPara]:

q = " ".join(terms)

return [p for p in [Link] if all([Link]() in [Link]() for t in terms)]

def as_evidence(self) -> List[Dict]:

return [p.__dict__ for p in [Link]]

Layout

config/

[Link]

src/

gen_plan.py # Orchestrator

indexing/

protocol_index.py

spec_index.py # stubbed for widths & spans

retriever/

[Link] # EvidencePack (per-plan)

generation/

generator_llm.py # Single-LLM client + candidates

validation/

grounding_validator.py # Python E-rules

rewriter/

[Link] # Surgical fixes

ranking/

[Link]

[Link] # CLI entry

config/[Link]

[multi_agent]

enabled = true

draft_candidates = 5

rank_top_k = 5
[validator]

forbid_performance_claims = true

enforce_width_ranges = true

[rewriter]

inject_missing_triggers = true

placeholder_timing = "(unspecified)"

[runtime]

workers = 4

[llm]

model = "gpt-4o-mini"

temperature_generator = 0.5

temperature_rewriter = 0.1

src/indexing/protocol_index.py

import re

from dataclasses import dataclass

from typing import List, Dict

@dataclass

class ProtocolPara:

id: str

text: str

numbers: List[int]

tokens_hex: List[str]
 timing_mentions: List[str]

class ProtocolIndex:

HEX_RE = [Link](r"0x[0-9A-Fa-f]+")

DIGITS_RE = [Link](r"\b\d+\b")

TIMING_WORDS =
{"cycle","clock","latency","period","baud","MHz","kHz","next"}

def __init__(self, paragraphs: List[str], prefix="txt:protocol"):

[Link]: List[ProtocolPara] = []

for i, txt in enumerate(paragraphs, 1):

pid = f"{prefix}:para{i}"

hexes = self.HEX_RE.findall(txt)

nums = list(map(int, self.DIGITS_RE.findall(txt)))

timing = [w for w in self.TIMING_WORDS if [Link](rf"\b{w}\b", txt,

re.I)]

[Link](ProtocolPara(pid, txt, nums, hexes, timing))

def query(self, terms: List[str]) -> List[ProtocolPara]:

return [p for p in [Link] if all([Link]() in [Link]() for t in terms)]

def as_evidence(self) -> List[Dict]:

return [p.__dict__ for p in [Link]]

src/indexing/spec_index.py (stub)

from typing import Dict, List

class SpecIndex:

def __init__(self, signal_rows: Dict[str, str], signal_widths: Dict[str, int]):

[Link] = signal_rows # e.g., {"int_rd_data": "row text ..."}

[Link] = signal_widths # e.g., {"int_rd_data": 8, "int_address":

16}

def lookup_signal_spans(self, signal: str):

# ultra-minimal numbers extractor

text = [Link](signal, "")

nums = [int(t) for t in [Link]() if [Link]()]

return {"ids": [f"tbl:signals:{signal}"], "numbers": nums}

def widths_for_referenced(self, signals: List[str]) -> Dict[str, int]:

return {s: [Link](s) for s in signals if s in [Link]}

def signal_dict(self) -> List[str]:

return list([Link]())

src/retriever/[Link] (per-plan EvidencePack)

from typing import Dict, List

from ..indexing.protocol_index import ProtocolIndex

class Retriever:

def __init__(self, spec_index, protocol_index: ProtocolIndex):

[Link] = spec_index

[Link] = protocol_index
 def build_pack(self, signal: str, referenced_signals: List[str],
protocol_terms: List[str]):

spans = [Link].lookup_signal_spans(signal)

proto = [Link](protocol_terms) if protocol_terms else []

allowed_numbers = sorted(set(

spans["numbers"] + [n for p in proto for n in [Link]]

))

evidence_span_ids = spans["ids"] + [[Link] for p in proto]

widths = [Link].widths_for_referenced(referenced_signals)

return {

"signal": signal,

"allowed_numbers": allowed_numbers,

"evidence_span_ids": evidence_span_ids,

"widths": widths,

"signal_dict": [Link].signal_dict()

src/generation/generator_llm.py

from typing import List, Dict, Any, Optional

import json

GEN_SYS = """You are a hardware spec planner.

Output STRICT JSON only, matching this schema:

"signal": str,

"trigger": str,

"effects": [str, ...],

"timing": "(unspecified)" | int,

"ranges": { signal: { "min": int, "max": int } | "(unspecified)" },

"evidence_span_ids": [str, ...]

Rules:

- Use ONLY numbers contained in allowed_numbers or implied by bit widths.

- If unknown, write "(unspecified)" (do NOT invent).

- Do NOT claim performance (MHz/baud) unless evidence explicitly mentions

it (with numbers).

- Prefer given relations (if any).

Return JSON ONLY.

"""

GEN_USER_TMPL = """TASK: Draft {k} JSON plan candidates for signal:

{signal}

CONTEXT:

- allowed_numbers: {allowed_numbers}

- widths: {widths}

- evidence_span_ids: {evidence_span_ids}

HINTS:

- If mentioning int_rd_data, ensure it is produced by int_read.

- Next-cycle timing only if evidence mentions it.

"""

class SingleLLMClient:

def __init__(self, model: str, temperature: float = 0.5, call_fn=None):

[Link] = model
 [Link] = temperature

# call_fn(system_prompt, user_prompt, temperature) -> str

self.call_fn = call_fn or self._fake_call

def _fake_call(self, system_prompt: str, user_prompt: str, temperature:

float) -> str:

# Minimal deterministic mock

plan = {

"signal": "int_read",

"trigger": "int_read",

"effects": ["int_rd_data"],

"timing": "(unspecified)",

"ranges": {"int_rd_data": {"min": 0, "max": 255}},

"evidence_span_ids": ["tbl:signals:int_read"]

return [Link]([plan]) # return a JSON array of plans

def generate_candidates(self, signal: str, evidence_pack: Dict[str, Any],

k=5) -> List[Dict[str, Any]]:

user = GEN_USER_TMPL.format(

k=k,

signal=signal,

allowed_numbers=evidence_pack["allowed_numbers"],

widths=evidence_pack["widths"],

evidence_span_ids=evidence_pack["evidence_span_ids"],

raw = self.call_fn(GEN_SYS, user, [Link])

 # Expect either single or list

try:

data = [Link](raw)

if isinstance(data, dict): data = [data]

return data[:k]

except Exception:

return []

Plug your actual OpenAI call where _fake_call is (keeping system_prompt,

user_prompt, temperature).

src/validation/grounding_validator.py (Python E-rules)

from typing import Dict, Any, List, Tuple

class GroundingValidator:

def __init__(self, policy: Dict[str, Any]):

[Link] = policy

def _width_bounds(self, w: int) -> Tuple[int,int]:

return (0, (1<<w)-1)

def _within_width(self, num: int, widths: Dict[str,int]) -> bool:

for _, w in [Link]():

lo, hi = self._width_bounds(w)

if lo <= num <= hi:

return True

return False
 def _collect_numbers(self, plan: Dict[str, Any]) -> List[int]:

nums = []

# timing can be int

if isinstance([Link]("timing"), int): [Link](plan["timing"])

# ranges min/max

for v in ([Link]("ranges") or {}).values():

if isinstance(v, dict):

if isinstance([Link]("min"), int): [Link](v["min"])

if isinstance([Link]("max"), int): [Link](v["max"])

return nums

def validate(self, plan: Dict[str, Any], evidence: Dict[str, Any]) -> Dict[str,
Any]:

flags: List[str] = []

allowed = set([Link]("allowed_numbers", []))

widths = [Link]("widths", {})

sigdict = set([Link]("signal_dict", []))

# E1: evidence required

if not [Link]("evidence_span_ids"):

[Link]("E1: missing evidence_span_ids")

# E3: signals must exist

for s in [[Link]("signal"), [Link]("trigger")] + [Link]("effects", []):

if s and s not in sigdict: [Link](f"E3: unknown signal {s}")

# E2: numeric purity

 for n in self._collect_numbers(plan):

if (n not in allowed) and (not self._within_width(n, widths)):

[Link](f"E2: number {n} not allowed")

# E4: performance ban (very simple textual guard)

if [Link]("forbid_performance_claims", True):

joined = str(plan)

if any(w in joined for w in ["MHz","kHz","throughput","baud"]):

if not any("MHz" in e for e in [Link]("evidence_span_ids", [])):

[Link]("E4: performance claim without cited evidence")

# E5: next-cycle timing only with evidence (simple proxy rule)

if [Link]("timing") == 1:

if not any("next" in eid or "int_rd_data" in eid for eid in

[Link]("evidence_span_ids", [])):

[Link]("E5: next-cycle timing lacks evidence")

status = "ok" if not flags else "flagged"

return {"status": status, "plan": plan, "flags": flags}

src/rewriter/[Link]

from typing import Dict, Any, List

class PlanRewriter:

def __init__(self, placeholder_timing="(unspecified)",

inject_missing_triggers=True):

self.placeholder_timing = placeholder_timing
 self.inject_missing_triggers = inject_missing_triggers

def _timing_demote(self, plan: Dict[str, Any]) -> None:

plan["timing"] = self.placeholder_timing

def _inject_trigger_for_rd(self, plan: Dict[str, Any]) -> None:

if "int_rd_data" in [Link]("effects", []) and [Link]("trigger") !=

"int_read":

plan["trigger"] = "int_read"

def _normalize_ranges(self, plan: Dict[str, Any], widths: Dict[str,int]) ->

None:

ranges = [Link]("ranges", {})

for sig, w in [Link]():

if sig in ranges:

ranges[sig] = {"min": 0, "max": (1<<w)-1}

def rewrite(self, validator_result: Dict[str, Any], evidence: Dict[str, Any]) ->

Dict[str, Any]:

plan = dict(validator_result["plan"])

flags = validator_result["flags"]

if any([Link]("E5") for f in flags): self._timing_demote(plan)

if self.inject_missing_triggers: self._inject_trigger_for_rd(plan)

if any([Link]("E2") for f in flags): self._normalize_ranges(plan,

[Link]("widths", {}))

# scrub perf words

 joined = str(plan)

if any(w in joined for w in ["MHz","kHz","throughput","baud"]):

# simplest scrub: demote timing text and remove perf-y fields if

present

plan["timing"] = self.placeholder_timing

# ensure evidence list exists

[Link]("evidence_span_ids", [Link]("evidence_span_ids",
[]))

return plan

src/ranking/[Link]

from typing import List, Dict, Any

def select(plans: List[Dict[str, Any]], top_k: int = 5) -> List[Dict[str, Any]]:

# naive diversity: prefer different timing and effects shapes

key = lambda p: ([Link]("timing"), tuple(sorted([Link]("effects", []))))

uniq = {}

for p in plans:

[Link](key(p), p)

if len(uniq) >= top_k: break

return list([Link]())[:top_k]

src/gen_plan.py (orchestrator)

from typing import List, Dict, Any

from .[Link] import Retriever

from .generation.generator_llm import SingleLLMClient

from .validation.grounding_validator import GroundingValidator

from .[Link] import PlanRewriter

from .ranking import ranker

class Pipeline:

def __init__(self, retriever: Retriever, llm: SingleLLMClient, cfg: Dict[str,

Any]):

[Link] = retriever

[Link] = llm

[Link] = cfg

[Link] = GroundingValidator(policy={

"forbid_performance_claims": [Link]["validator"]
["forbid_performance_claims"]

})

[Link] = PlanRewriter(

placeholder_timing=[Link]["rewriter"]["placeholder_timing"],

inject_missing_triggers=[Link]["rewriter"]["inject_missing_triggers"]

def process_signal(self, signal: str, referenced_signals: List[str],

protocol_terms: List[str]) -> List[Dict[str, Any]]:

# 1) Build per-plan evidence

pack = [Link].build_pack(signal, referenced_signals,

protocol_terms)

# 2) Generate k JSON candidates

k = [Link]["multi_agent"]["draft_candidates"]

drafts = [Link].generate_candidates(signal, pack, k=k)

 # 3) Validate deterministically

judged = [[Link](d, pack) for d in drafts]

ok = [j["plan"] for j in judged if j["status"] == "ok"]

flagged = [j for j in judged if j["status"] == "flagged"]

# 4) Rewrite flagged → repair

repaired = [[Link](j, pack) for j in flagged]

# 5) Re-validate repaired

repaired_ok = [[Link](p, pack)["plan"] for p in repaired

if [Link](p, pack)["status"] == "ok"]

# 6) Rank & select

finalists = [Link](ok + repaired_ok, top_k=[Link]["multi_agent"]

["rank_top_k"])

# 7) Guarantee coverage (optional top-up)

if len(finalists) < [Link]["multi_agent"]["rank_top_k"]:

# schema-only retry could be added here (not shown for brevity)

pass

return finalists

[Link] (wire it up)

import toml

from [Link].spec_index import SpecIndex

from [Link].protocol_index import ProtocolIndex

from [Link] import Retriever

from [Link].generator_llm import SingleLLMClient

from src.gen_plan import Pipeline

def openai_call(system_prompt, user_prompt, temperature):

# Replace with your real OpenAI call (pseudo):

# client = OpenAI()

# resp = [Link](

# model="gpt-4o-mini",

# messages=[{"role":"system","content":system_prompt},

# {"role":"user","content":user_prompt}],

# temperature=temperature,

#)

# return [Link][0].[Link]

raise NotImplementedError("Wire your OpenAI call here")

def load_cfg(path="config/[Link]"):

return [Link](path)

if __name__ == "__main__":

cfg = load_cfg()

# Minimal demo indices

signal_rows = {

"int_read": "Read strobe; next cycle data valid.",

"int_rd_data": "8-bit data output.",

 "int_write": "Write strobe; same cycle data accept.",

"int_wr_data": "8-bit data input.",

"int_address": "16-bit address bus."

widths = {"int_rd_data": 8, "int_wr_data": 8, "int_address": 16}

spec = SpecIndex(signal_rows, widths)

protocol = ProtocolIndex([

"Controller acknowledges with 0x5A and may respond next cycle.",

"CR LF terminators for text mode."

])

retriever = Retriever(spec, protocol)

llm = SingleLLMClient(

model=cfg["llm"]["model"],

temperature=cfg["llm"]["temperature_generator"],

call_fn=openai_call # ← plug your real call

pipe = Pipeline(retriever, llm, cfg)

# Example run per signal

results = pipe.process_signal(

signal="int_read",

referenced_signals=["int_rd_data", "int_address"],

protocol_terms=["next","cycle"]

)
print("FINAL PLANS:")

for i, p in enumerate(results, 1):

print(f"[{i}] {p}")