feat: TransportPacket canonical cutover — Pack 3 tools + CI guard#108
feat: TransportPacket canonical cutover — Pack 3 tools + CI guard#108cryptoxdog wants to merge 2 commits into
Conversation
Introduce TransportPacket as the canonical immutable data container, replacing PacketEnvelope which is now a deprecated compatibility wrapper. Core changes: - l9_core/models.py: TransportPacket (dataclass, frozen, sha256 hash) with PacketEnvelope kept as deprecated shim - engine/packet/bridge.py: migrated to TransportPacket API - engine/arbitration/schema.py: added DecisionPolicy, HardConstraint, PolicyWeights, PolicyThresholds models - engine/arbitration/engine.py: fixed imports and type hints CI enforcement: - .pre-commit-config.yaml: terminology-guard regex targets active PacketEnvelope usage (imports/constructor calls) without false positives on deprecation documentation Tools (Pack 3 — declarative audit): - tools/audit_harness.py: refactored from subprocess to importlib - tools/validate_domain.py: domain spec validator - tools/domain_extractor.py: signal extractor - tools/l9_meta_injector.py: metadata injector - tools/l9_template_manifest.yaml: manifest definition Pre-existing fixes (broken on main): - DomainSpecLoader → DomainPackLoader import across 14 test files - DomainPackLoader constructor: domains_dir → config_path - tests/conftest.py: added missing queryentity field to minimal spec - test_wave6_dormant_features: fixed LLM test match regex - test_injection/test_sync_generator: fixed RUF059 unused unpacked vars Made-with: Cursor
|
❌ PR Too Large 📋 Best Practices for Large Changes
🚫 This PR is blocked from merging until size limits are met. |
📝 WalkthroughWalkthroughThis pull request refactors the packet model and bridge layer from Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ce02811243
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| print(" ❌ Missing spec features (--strict mode)") | ||
| def main() -> int: | ||
| parser = argparse.ArgumentParser(description="Strict cutover audit harness") | ||
| parser.add_argument("repo_root", type=Path) |
There was a problem hiding this comment.
This change makes repo_root mandatory, but existing invocations in this repo still call python tools/audit_harness.py without positional args (for example the pre-push hook in .pre-commit-config.yaml and Makefile harness targets). In those paths the script now exits with argparse error code 2, so the audit harness can no longer run in CI/local workflow until callers are updated or a default (e.g. cwd) is restored.
Useful? React with 👍 / 👎.
| msg = f"Cannot create module spec for {file_path}" | ||
| raise ImportError(msg) | ||
| mod = importlib.util.module_from_spec(spec) | ||
| spec.loader.exec_module(mod) |
There was a problem hiding this comment.
_load_tool_module executes modules directly but never inserts them into sys.modules first. Tools that define dataclasses (e.g., tools/contract_scanner.py's Violation) can fail at import time because decorators resolve module metadata through sys.modules, causing the harness to crash before any audit step completes.
Useful? React with 👍 / 👎.
| help="Skip contract wiring verification", | ||
| def _step_contract_scanner(tools_dir: Path, repo_root: Path, out_dir: Path) -> tuple[dict[str, Any], int, int]: | ||
| cs = _load_tool_module("contract_scanner", tools_dir) | ||
| violations: list[Any] = cs.scan_repo(root=repo_root, explicit_paths=[]) |
There was a problem hiding this comment.
This step calls cs.scan_repo(...), but tools/contract_scanner.py does not define a scan_repo function, so the harness will raise AttributeError when reaching this line (once module-loading issues are fixed). The previous implementation invoked the scanner via its script entrypoint, so this new direct-call path is currently broken.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 15
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
tests/unit/test_wave6_dormant_features.py (1)
620-628:⚠️ Potential issue | 🟡 MinorNote: assertion depends on
openaibeing importable.
ValidatedLLMClient._call()re-raises the underlyingRuntimeErrortext as theFeatureNotEnabledmessage. That text contains"OPENAI_API_KEY"only on the missing-env-var branch of_ensure_client(); on the missing-openai-package branch the message is"openai package is required...", which would fail this match. Sinceopenai==1.0.0is inrequirements.txtthis should be fine in CI, but the assertion is order-of-failure dependent.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_wave6_dormant_features.py` around lines 620 - 628, The test test_llm_client_raises_feature_not_enabled relies on the exact error text raised by ValidatedLLMClient._call/_ensure_client and can fail depending on whether the openai package is missing or the env var is missing; update the assertion to accept both possibilities by checking that a FeatureNotEnabled is raised and that the exception message contains either "OPENAI_API_KEY" or "openai package is required" (or use a regex like "(OPENAI_API_KEY|openai package is required)"), referencing ValidatedLLMClient._call and the FeatureNotEnabled exception so the test is robust regardless of which underlying branch of _ensure_client fails.tests/unit/test_domain_loader.py (1)
57-64:⚠️ Potential issue | 🟠 MajorFix test directory structure: create
broken/spec.yamlsubdirectory instead ofbroken.yaml.The loader resolves
load_domain("broken")toconfig_path/broken/spec.yaml, but the test writestmp_path/broken.yaml. The file will not be found at the expected path, raisingDomainNotFoundErrorbefore YAML parsing is attempted. The test passes by accident due to catching a genericException.Create the file at the correct location:
Corrected test structure
def test_malformed_yaml_raises(tmp_path): broken_dir = tmp_path / "broken" broken_dir.mkdir() spec_file = broken_dir / "spec.yaml" spec_file.write_text("domain:\n id: [unclosed") from engine.config.loader import DomainPackLoader loader = DomainPackLoader(config_path=str(tmp_path)) with pytest.raises(Exception): loader.load_domain("broken")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_domain_loader.py` around lines 57 - 64, The test test_malformed_yaml_raises writes the YAML to tmp_path/broken.yaml but DomainPackLoader.load_domain("broken") expects a directory tmp_path/broken/spec.yaml, so the test accidentally raises DomainNotFoundError; fix by creating a "broken" directory and writing the malformed YAML to "spec.yaml" inside it before instantiating DomainPackLoader and calling loader.load_domain("broken") (update references in the test_malformed_yaml_raises function around DomainPackLoader and loader.load_domain to use the corrected tmp_path/broken/spec.yaml layout).
🧹 Nitpick comments (11)
tools/l9_meta_injector.py (1)
67-70: Nit: simpler counter idiom.
updated += 1 if inject(path) else 0can be flattened, and the early-skip pattern reads better than the nestedif.♻️ Optional cleanup
- for path in root.rglob("*"): - if _should_process(path): - updated += 1 if inject(path) else 0 + for path in root.rglob("*"): + if not _should_process(path): + continue + if inject(path): + updated += 1🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tools/l9_meta_injector.py` around lines 67 - 70, The loop uses a nested conditional to increment updated: replace `if _should_process(path): updated += 1 if inject(path) else 0` with a clearer idiom—either skip early with `if not _should_process(path): continue` and then `if inject(path): updated += 1`, or compute updated with a generator expression (e.g., sum(1 for p in root.rglob("*") if _should_process(p) and inject(p))); reference the loop over root.rglob("*"), the _should_process(path) predicate, and the inject(path) call to locate and apply the change.tools/domain_extractor.py (1)
36-39: Consider also handlingOSErroronread_text.Recursive scans across a developer/CI workspace can encounter unreadable files (permission denied, broken symlinks, special files). Currently only
UnicodeDecodeErroris caught, so an unrelatedOSErrorwill abort the whole signal extraction.- try: - text = path.read_text(encoding="utf-8") - except UnicodeDecodeError: - continue + try: + text = path.read_text(encoding="utf-8") + except (UnicodeDecodeError, OSError): + continue🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tools/domain_extractor.py` around lines 36 - 39, The try/except around path.read_text only handles UnicodeDecodeError and can let OSError (e.g., permission errors, broken symlinks) abort the scan; update the exception handling in the try block that calls path.read_text to also catch OSError (or a tuple like (UnicodeDecodeError, OSError)) and continue on those failures so recursive scans skip unreadable files instead of crashing.tools/audit_harness.py (1)
23-35: Register the loaded module insys.modulesbefore executing it.The current code skips the standard step of inserting
sys.modules[name] = modbefore executing the module. This breaks: (a) any tool module that performs an internal self-reference (e.g.,import contract_scannerinsidecontract_scanner), (b)dataclasses.asdict/pickleround-trips when types are looked up by qualified name, and (c) repeated loads where a stale prior instance shadows the new one.Suggested fix
mod = importlib.util.module_from_spec(spec) + sys.modules[name] = mod spec.loader.exec_module(mod) return modAdd
import sysat the top. This matches the canonical importlib pattern in the official Python documentation.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tools/audit_harness.py` around lines 23 - 35, The _load_tool_module function must register the newly created module in sys.modules before executing it to support intra-module imports, pickling/dataclass lookups, and repeated reloads: add import sys to the file and set sys.modules[name] = mod (using the same name passed to _load_tool_module) immediately after creating mod (before calling spec.loader.exec_module(mod)); keep the existing spec/loader checks and exec_module call but ensure the module is inserted into sys.modules first.tests/unit/test_compliance_checker.py (1)
27-31: Broadexcept Exception→pytest.skipmasks genuine regressions.These compliance tests now silently skip on any exception during loading. Per the file's purpose (and the compliance-test guideline: "verify prohibited factors are blocked at compile time"), an unexpected loader/schema failure should fail loudly, not be hidden behind a "schema mismatch on main" skip. Recommend narrowing to the specific exception types you actually expect (e.g.,
pydantic.ValidationError,FileNotFoundError), or marking withpytest.mark.xfail(reason=..., strict=False)so the failure mode is visible.As per coding guidelines: "HIGH: bare except: → error handling violation (ERR-001). All exceptions must be caught explicitly" —
except Exceptionis technically explicit, but the practice of converting arbitrary failures to skips is the same anti-pattern this rule targets.Also applies to: 54-58
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_compliance_checker.py` around lines 27 - 31, The current try/except around DomainPackLoader(...) and loader.load_domain("plasticos") masks all errors by calling pytest.skip; replace the broad except Exception with explicit exception handling for expected failures (e.g., except (pydantic.ValidationError, FileNotFoundError): pytest.skip(...)) or instead mark the test as xfail (pytest.mark.xfail(..., strict=False)) so unexpected loader/schema failures surface as test failures; update the same pattern at the second occurrence (lines referencing DomainPackLoader/load_domain) accordingly..pre-commit-config.yaml (1)
109-109: Stale comment:PacketEnvelopeis no longer the canonical term.The header comment still advertises
PacketEnvelopeas canonical, contradicting both the new pattern below and the failure message at Line 124.📝 Proposed fix
- # Terminology guard (canonical terms: PacketEnvelope, list[T], T | None) + # Terminology guard (canonical terms: TransportPacket, list[T], T | None)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In @.pre-commit-config.yaml at line 109, The header comment labeled "Terminology guard" still advertises PacketEnvelope as the canonical term; update that comment to match the actual canonical term used by the rule/failure message and the new pattern (remove or replace "PacketEnvelope" and ensure the canonical examples such as "list[T], T | None" reflect the current policy), so the header aligns with the validation text referenced in the check that mentions the new canonical term.tests/security/test_injection.py (1)
39-46: Optional: strengthen the injection assertion.Currently the test only asserts that the evil string is absent from the Cypher text. Discarding
_paramsmeans a regression that drops the value entirely (rather than parameterizing it) would still pass. Consider asserting that the evil string is present inparamsto confirm it was actually parameterized, not silently dropped.♻️ Suggested tightening
- cypher, _params = gen.generate_sync_query(ep, [evil_record]) + cypher, params = gen.generate_sync_query(ep, [evil_record]) assert "DETACH DELETE" not in cypher assert "'; MATCH" not in cypher + # Value must be parameterized, not interpolated or dropped + assert any(evil_record["id"] in str(v) for v in _flatten_params(params))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/security/test_injection.py` around lines 39 - 46, Update the test to ensure the malicious payload is parameterized instead of dropped: after calling DomainPackLoader(...)->load_domain("plasticos") and SyncGenerator(spec).generate_sync_query(ep, [evil_record]) capture both cypher and params (the returned _params) and assert that the raw evil string does not appear in cypher but does appear in the params collection (e.g., params values contain "'; MATCH (n) DETACH DELETE n RETURN '1"), referencing the generate_sync_query return value to confirm parameterization rather than omission.reports/GMP-Report-131-Fix-DomainPackLoader-Import-And-Constructor.md (1)
7-7: Add language identifiers to fenced code blocks.
markdownlint-cli2flagsMD040on the fenced blocks at lines 7, 29, and 156. Suggest tagging each with an explicit language (e.g.,text,yaml) to satisfy the lint rule.Also applies to: 29-29, 156-156
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@reports/GMP-Report-131-Fix-DomainPackLoader-Import-And-Constructor.md` at line 7, The report contains fenced code blocks missing language identifiers (MD040) at the three fenced blocks referenced; update each opening fence (the three occurrences currently as ``` ) to include an explicit language tag like ```text or ```yaml depending on the content so markdownlint-cli2 MD040 is satisfied; search for the three untagged fences near the blocks mentioned and replace them with appropriate language-labeled fences (e.g., change ``` to ```text for plain output or ```yaml for YAML snippets) ensuring consistency across the blocks.engine/arbitration/schema.py (2)
37-40: ConstrainHardConstraint.metricto known input fields.
metric: strlets callers construct constraints whose metric does not exist onArbitrationInput, in which caseengine.resolvewill raiseAttributeErrorfromgetattr(data, constraint.metric). Tightening this to aLiteralof the valid numeric fields gives Pydantic-time validation and matches the_evaluate(actual: float, ..., expected: float)signature inengine/arbitration/engine.py. Note thatcompliance_passisbool— including it here would silently break the float contract of_evaluate, so it should remain excluded.♻️ Proposed change
class HardConstraint(BaseModel): - metric: str + metric: Literal["revenue", "margin", "risk", "capacity"] operator: Literal["eq", "lt", "lte", "gt", "gte"] value: float🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@engine/arbitration/schema.py` around lines 37 - 40, Change HardConstraint.metric from a free-form str to a Literal union of the numeric fields defined on ArbitrationInput so invalid metric names are rejected by Pydantic at model creation; update the Literal to list only the numeric fields (exclude compliance_pass because it is bool) and ensure any references in engine.resolve and _evaluate(actual: float, ..., expected: float) still assume the metric value is numeric so no runtime getattr(AttributeError) occurs for unknown metrics.
37-60: Inconsistentextrapolicy across schema models.
ArbitrationInput/ArbitrationResultusemodel_config = ConfigDict(extra="forbid"), but the newHardConstraint,PolicyWeights,PolicyThresholds, andDecisionPolicymodels accept arbitrary extra keys. This makes the policy schema silently tolerant of typos (e.g.,approve_threshvsapprove_threshold) which would fall back to defaults at runtime. Recommend matching the existing strict-extra policy on the new models.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@engine/arbitration/schema.py` around lines 37 - 60, The new policy-related Pydantic models (HardConstraint, PolicyWeights, PolicyThresholds, DecisionPolicy) currently allow arbitrary extra fields which hides typos; update each of these classes to enforce the same strict extra policy as ArbitrationInput/ArbitrationResult by adding model_config = ConfigDict(extra="forbid") to HardConstraint, PolicyWeights, PolicyThresholds, and DecisionPolicy (and import ConfigDict if not already present) so unknown keys raise validation errors instead of being silently ignored.l9_core/models.py (1)
57-73: Deprecated shim emits no deprecation signal.The class docstring labels
PacketEnvelopeas deprecated, but neither the class norfrom_transport_packet/to_transport_packetraiseDeprecationWarning(or usetyping.deprecatedfrom PEP 702 /typing_extensions.deprecated). Without a runtime signal, callers will keep using the shim indefinitely and the cutover will stall. Note also thatfrom_transport_packetdiscardspacket_id,content_hash, and the authority fields, andto_transport_packetmints a freshpacket_id— so the conversion is lossy in both directions; callers who treat it as a round-trip will silently lose identity.♻️ Add a runtime deprecation signal
+import warnings + `@dataclass`(frozen=True) class PacketEnvelope: """Deprecated compatibility wrapper. This type must not be used as canonical runtime truth. """ packet_type: str action: str payload: dict[str, Any] + def __post_init__(self) -> None: + warnings.warn( + "PacketEnvelope is deprecated; use l9_core.models.TransportPacket.", + DeprecationWarning, + stacklevel=2, + ) + `@classmethod` def from_transport_packet(cls, packet: TransportPacket) -> PacketEnvelope:🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@l9_core/models.py` around lines 57 - 73, PacketEnvelope is documented as deprecated but emits no runtime deprecation signal and its conversions drop critical fields (packet_id, content_hash, authority) and mint a new packet_id, causing silent data loss; update PacketEnvelope, its class methods from_transport_packet and to_transport_packet to emit a DeprecationWarning (use warnings.warn(..., DeprecationWarning) or typing_extensions.deprecated if available) when called, and change the conversion logic to preserve and round-trip TransportPacket's packet_id, content_hash, and authority fields (or explicitly document and include them on the payload) instead of discarding them or creating a fresh id; ensure warnings reference PacketEnvelope/from_transport_packet/to_transport_packet and that to_transport_packet reuses the original packet_id when present or stores it in payload to avoid silent identity loss.tests/unit/test_packet_bridge.py (1)
39-51: Test name overstates the assertion.
test_packet_bridge_derive_preserves_actiononly checksderived.action == "outcome_event"(the hard-coded action set byoutcome_packet) and thatpacket_iddiffers — it does not verify any lineage is preserved (there is none onTransportPacket). Either rename totest_packet_bridge_outcome_packet_sets_action_and_new_idto match what's actually tested, or — when lineage is added back toTransportPacketper the comment inl9_core/models.py— extend this case to assertderived.parent_id == root.packet_idetc.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_packet_bridge.py` around lines 39 - 51, The test name test_packet_bridge_derive_preserves_action is misleading because it only asserts that PacketBridge.outcome_packet sets action to "outcome_event" and that the derived.packet_id differs from root.packet_id; update the test to either rename it to test_packet_bridge_outcome_packet_sets_action_and_new_id to match the current assertions, or if/when TransportPacket lineage is restored (per l9_core/models.py) extend the assertions to check lineage (e.g., assert derived.parent_id == root.packet_id) and any other parent/child fields; locate the test and the PacketBridge.inflate_ingress / PacketBridge.outcome_packet calls to apply the change.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In @.pre-commit-config.yaml:
- Around line 117-119: The grep hook that scans for "PacketEnvelope" and other
patterns should be changed so it does not flag the module that defines/exports
PacketEnvelope or the contract test files; modify the loop that iterates
patterns (the for pattern in "import.*PacketEnvelope|PacketEnvelope\\(" ... ) to
skip the package that defines PacketEnvelope and the contract-tests directory by
adding exclusions (e.g., add --exclude-dir for the packet package and contract
tests or maintain an explicit exemption list of filenames like the defining
module, its __init__ export, and the contract test files) so only other files
are prevented from redefining/using PacketEnvelope. Ensure the adjusted git
diff/grep pipeline still searches the intended engine/tests scope but filters
out those exempted locations.
In `@engine/packet/bridge.py`:
- Around line 9-13: inflate_ingress (and attach_entity_semantics) currently
flattens routing/identity keys into the payload which overwrites caller keys and
erodes the typed envelope; change these functions to stop merging
tenant_id/actor/entity_type/canonical_entity_type into payload and instead set
them on TransportPacket's dedicated envelope/metadata fields (e.g.,
TransportPacket.create(..., metadata={...}) or explicit
TransportPacket.tenant_id/actor fields), or alternatively prefix them (e.g.,
_meta.tenant_id) if you cannot change the TransportPacket shape; ensure you do
not perform {**payload, "tenant_id": ...} merging so caller keys aren't
overwritten and the envelope retains typed routing fields.
In `@l9_core/models.py`:
- Around line 51-54: The derive method currently mints a fresh TransportPacket
and drops provenance; update derive (in TransportPacket / models.py) to preserve
and extend lineage instead of calling TransportPacket.create(...) with only
action/payload: set parent_id=self.packet_id, set root_id=self.root_id or
self.packet_id if root_id is None, set generation=(self.generation or 0)+1, copy
delegation_chain=self.delegation_chain, and set hop_trace=self.hop_trace + [new
HopEntry(...)] (create a HopEntry representing this derivation); then call
TransportPacket.create or its constructor with these explicit lineage fields so
downstream code (bridge.py functions attach_entity_semantics, decision_packet,
outcome_packet) retains parent_id/root_id/generation/hop_trace/delegation_chain.
Ensure field names match the TransportPacket schema and use existing HopEntry
type for the appended entry.
- Around line 16-45: The content hash currently includes packet_id, which makes
content-addressing ineffective; update TransportPacket so content_hash excludes
packet_id by computing _stable_hash over only the actual content fields (action,
payload, runtime_authority, routing_authority, transport_authority) instead of
including packet_id; modify TransportPacket.__post_init__ to call _stable_hash
with a dict that omits "packet_id" (and ensure any helper like
TransportPacket.create or .derive continues to set packet_id separately),
keeping the field name content_hash and using the existing _stable_hash
function.
- Around line 57-73: Remove the duplicate dataclass PacketEnvelope in
l9_core.models: eliminate the deprecated PacketEnvelope class (including its
from_transport_packet and to_transport_packet methods) so the canonical Pydantic
PacketEnvelope in engine.packet.packet_envelope.py remains the single source of
truth; alternatively, if a compatibility shim is required during cutover, move
the conversion helpers (from_transport_packet, to_transport_packet) into
engine.packet.packet_envelope.py or add a thin wrapper there and re-export it
for backward compatibility, ensuring all imports reference
engine.packet.packet_envelope.PacketEnvelope and no class definition remains in
l9_core.models.
In `@tests/unit/test_outcomes.py`:
- Around line 10-14: The test passes a spec file to DomainPackLoader via the
SPEC_PATH variable which sets config_path to a YAML file instead of the required
domains base directory; change SPEC_PATH to point to the domains directory
(derive it relative to the test file instead of relying on CWD) and pass that to
DomainPackLoader(config_path=...) so the loader's self._base_path is a
directory; update the test to mirror other tests by computing the base "domains"
directory from __file__ (e.g., using Path(__file__).parents[...] joined with
"domains") and use that when constructing DomainPackLoader.
In `@tests/unit/test_packet_bridge.py`:
- Around line 16-21: The test test_transport_packet_hash_changes_with_payload is
tautological because TransportPacket.create() generates a fresh packet_id (used
in __post_init__/_stable_hash), so different UUIDs guarantee different
content_hashes; update the test to pin a deterministic packet_id when creating
the two packets (use the same explicit packet_id for both creations or construct
TransportPacket with a fixed UUID) so the only difference between p1 and p2 is
the payload, then assert content_hash differs; reference TransportPacket.create,
packet_id, content_hash, and the _stable_hash/__post_init__ behavior when making
the change.
In `@tools/audit_harness.py`:
- Around line 109-119: The CLI currently requires the positional repo_root and
CI invokes the harness without it; make repo_root optional and default to the
current repository root (safe choice: Path.cwd().resolve()) so CI passes. Update
parser.add_argument("repo_root", ...) to accept an optional value (e.g.,
nargs="?" or provide default=None) and then change the assignment to use a
fallback: set repo_root = (args.repo_root or Path.cwd()).resolve(); leave
tools_dir and out_dir logic unchanged (variables: repo_root, tools_dir, out_dir;
function: main()).
In `@tools/domain_extractor.py`:
- Around line 12-17: The "Gate" token in SIGNALS (routing_authority) causes
false positives because it's a substring of "Gate_SDK" (runtime_authority) and
other identifiers; change the matching logic that currently does plain substring
checks to use word-boundary regex for tokens (e.g., match r"\bGate\b") and
ensure routing_authority matching explicitly excludes runtime_authority tokens
(or check runtime_authority first and skip routing for those occurrences) so
counts computed for the SIGNALS dict (routing_authority, runtime_authority,
canonical_transport, deprecated_compatibility) are mutually exclusive and avoid
double-counting.
In `@tools/l9_meta_injector.py`:
- Around line 2-6: The module docstring is ambiguous; update the sentence about
PacketEnvelope to clearly state the injector's contract: e.g., say that the
injector treats TransportPacket as canonical and will only mark files that
reference PacketEnvelope with compatibility_surface: true (instead of “not
normalize deprecated PacketEnvelope-first architecture as active truth”). Edit
the module docstring in tools/l9_meta_injector.py to use this clearer phrasing
and reference TransportPacket, PacketEnvelope, and compatibility_surface: true.
- Around line 52-58: The inject function currently can raise UnicodeDecodeError
or OSError and abort the run; wrap the file operations in a try/except around
path.read_text and path.write_text inside inject (while keeping the existing
_has_header, HEADER_TEMPLATE and _compatibility_hint usage) and catch
UnicodeDecodeError and OSError, log a warning (include path and exception) and
return False to skip the file so traversal in main can continue; ensure you only
swallow these expected file I/O errors and let unexpected exceptions propagate.
- Around line 13-21: TEXT_EXTENSIONS currently includes ".json" and ".md" which
makes HEADER_TEMPLATE (prefixed "#" lines) unsafe: remove ".json" from
TEXT_EXTENSIONS and either remove ".md" or make injection conditional by
branching on extension in the injector (use an HTML comment wrapper like "<!--
L9_META: ... -->" for ".md"); for ".json" avoid in-file mutation (emit a sidecar
file such as "<original>.l9meta" or "<original>.json.l9meta") instead. Update
the logic that uses TEXT_EXTENSIONS and HEADER_TEMPLATE so it selects the proper
per-extension header or sidecar behavior rather than blindly prepending
HEADER_TEMPLATE to all entries.
- Around line 61-73: main() currently walks root.rglob("*") with no exclusions
and always calls inject(path) which uses path.write_text, risking mass
destructive edits; update main() to (1) add an explicit --apply boolean flag
defaulting to False so runs are dry-run by default and only write when --apply
is provided, (2) build an exclusion set of well-known non-source dirs (e.g.,
".git", ".venv", "node_modules", "build", "dist", vendor/generated/test fixture
dirs) and skip any paths under those in the walk (use _should_process(path) or
enhance it to check the exclusion set), and (3) collect and print the list of
changed file paths (instead of just a count) when dry-run or after applying so
users can review edits; ensure inject(path) only calls path.write_text when the
--apply flag is True.
In `@tools/l9_template_manifest.yaml`:
- Around line 31-36: The manifest's token lists (forbidden_primary_signals /
required_signals / authorities) must be the single source of truth; update
tools/domain_extractor.py (SIGNALS["deprecated_compatibility"]) and
tools/validate_domain.py (FORBIDDEN_VALUES) to load and derive their token sets
from the YAML manifest (e.g.,
yaml.safe_load(Path("tools/l9_template_manifest.yaml").read_text())) instead of
hardcoding values so entries like "chassis.router" and the exact forbidden
tokens (including parentheses where present) stay in sync with
strict_fail_closed enforcement. Ensure both modules map the manifest keys to
their internal structures (SIGNALS[...] and FORBIDDEN_VALUES) and remove the
duplicated hardcoded lists.
In `@tools/validate_domain.py`:
- Around line 18-27: Replace the hardcoded FORBIDDEN_VALUES and
REQUIRED_AUTHORITY with values loaded at runtime from
tools/l9_template_manifest.yaml (derive forbidden_primary_signals and
authorities once at module init), update forbidden tokens to match the manifest
exactly (e.g., use "inflate_ingress(" and "deflate_egress(" rather than bare
identifiers) and update any occurrences at lines referenced; then modify
validate_file to consult the manifest's
deprecated_compatibility.packet_envelope.allowed_only_in (or equivalent
allowed_only_in entries) and skip the forbidden-token check when the file path
lives under any allowed context directories (e.g., compatibility_shims,
migration_docs, compatibility_tests, scanner_allowlists), ensuring behavior
mirrors the manifest-derived rules rather than unconditional flagging.
---
Outside diff comments:
In `@tests/unit/test_domain_loader.py`:
- Around line 57-64: The test test_malformed_yaml_raises writes the YAML to
tmp_path/broken.yaml but DomainPackLoader.load_domain("broken") expects a
directory tmp_path/broken/spec.yaml, so the test accidentally raises
DomainNotFoundError; fix by creating a "broken" directory and writing the
malformed YAML to "spec.yaml" inside it before instantiating DomainPackLoader
and calling loader.load_domain("broken") (update references in the
test_malformed_yaml_raises function around DomainPackLoader and
loader.load_domain to use the corrected tmp_path/broken/spec.yaml layout).
In `@tests/unit/test_wave6_dormant_features.py`:
- Around line 620-628: The test test_llm_client_raises_feature_not_enabled
relies on the exact error text raised by ValidatedLLMClient._call/_ensure_client
and can fail depending on whether the openai package is missing or the env var
is missing; update the assertion to accept both possibilities by checking that a
FeatureNotEnabled is raised and that the exception message contains either
"OPENAI_API_KEY" or "openai package is required" (or use a regex like
"(OPENAI_API_KEY|openai package is required)"), referencing
ValidatedLLMClient._call and the FeatureNotEnabled exception so the test is
robust regardless of which underlying branch of _ensure_client fails.
---
Nitpick comments:
In @.pre-commit-config.yaml:
- Line 109: The header comment labeled "Terminology guard" still advertises
PacketEnvelope as the canonical term; update that comment to match the actual
canonical term used by the rule/failure message and the new pattern (remove or
replace "PacketEnvelope" and ensure the canonical examples such as "list[T], T |
None" reflect the current policy), so the header aligns with the validation text
referenced in the check that mentions the new canonical term.
In `@engine/arbitration/schema.py`:
- Around line 37-40: Change HardConstraint.metric from a free-form str to a
Literal union of the numeric fields defined on ArbitrationInput so invalid
metric names are rejected by Pydantic at model creation; update the Literal to
list only the numeric fields (exclude compliance_pass because it is bool) and
ensure any references in engine.resolve and _evaluate(actual: float, ...,
expected: float) still assume the metric value is numeric so no runtime
getattr(AttributeError) occurs for unknown metrics.
- Around line 37-60: The new policy-related Pydantic models (HardConstraint,
PolicyWeights, PolicyThresholds, DecisionPolicy) currently allow arbitrary extra
fields which hides typos; update each of these classes to enforce the same
strict extra policy as ArbitrationInput/ArbitrationResult by adding model_config
= ConfigDict(extra="forbid") to HardConstraint, PolicyWeights, PolicyThresholds,
and DecisionPolicy (and import ConfigDict if not already present) so unknown
keys raise validation errors instead of being silently ignored.
In `@l9_core/models.py`:
- Around line 57-73: PacketEnvelope is documented as deprecated but emits no
runtime deprecation signal and its conversions drop critical fields (packet_id,
content_hash, authority) and mint a new packet_id, causing silent data loss;
update PacketEnvelope, its class methods from_transport_packet and
to_transport_packet to emit a DeprecationWarning (use warnings.warn(...,
DeprecationWarning) or typing_extensions.deprecated if available) when called,
and change the conversion logic to preserve and round-trip TransportPacket's
packet_id, content_hash, and authority fields (or explicitly document and
include them on the payload) instead of discarding them or creating a fresh id;
ensure warnings reference
PacketEnvelope/from_transport_packet/to_transport_packet and that
to_transport_packet reuses the original packet_id when present or stores it in
payload to avoid silent identity loss.
In `@reports/GMP-Report-131-Fix-DomainPackLoader-Import-And-Constructor.md`:
- Line 7: The report contains fenced code blocks missing language identifiers
(MD040) at the three fenced blocks referenced; update each opening fence (the
three occurrences currently as ``` ) to include an explicit language tag like
```text or ```yaml depending on the content so markdownlint-cli2 MD040 is
satisfied; search for the three untagged fences near the blocks mentioned and
replace them with appropriate language-labeled fences (e.g., change ``` to
```text for plain output or ```yaml for YAML snippets) ensuring consistency
across the blocks.
In `@tests/security/test_injection.py`:
- Around line 39-46: Update the test to ensure the malicious payload is
parameterized instead of dropped: after calling
DomainPackLoader(...)->load_domain("plasticos") and
SyncGenerator(spec).generate_sync_query(ep, [evil_record]) capture both cypher
and params (the returned _params) and assert that the raw evil string does not
appear in cypher but does appear in the params collection (e.g., params values
contain "'; MATCH (n) DETACH DELETE n RETURN '1"), referencing the
generate_sync_query return value to confirm parameterization rather than
omission.
In `@tests/unit/test_compliance_checker.py`:
- Around line 27-31: The current try/except around DomainPackLoader(...) and
loader.load_domain("plasticos") masks all errors by calling pytest.skip; replace
the broad except Exception with explicit exception handling for expected
failures (e.g., except (pydantic.ValidationError, FileNotFoundError):
pytest.skip(...)) or instead mark the test as xfail (pytest.mark.xfail(...,
strict=False)) so unexpected loader/schema failures surface as test failures;
update the same pattern at the second occurrence (lines referencing
DomainPackLoader/load_domain) accordingly.
In `@tests/unit/test_packet_bridge.py`:
- Around line 39-51: The test name test_packet_bridge_derive_preserves_action is
misleading because it only asserts that PacketBridge.outcome_packet sets action
to "outcome_event" and that the derived.packet_id differs from root.packet_id;
update the test to either rename it to
test_packet_bridge_outcome_packet_sets_action_and_new_id to match the current
assertions, or if/when TransportPacket lineage is restored (per
l9_core/models.py) extend the assertions to check lineage (e.g., assert
derived.parent_id == root.packet_id) and any other parent/child fields; locate
the test and the PacketBridge.inflate_ingress / PacketBridge.outcome_packet
calls to apply the change.
In `@tools/audit_harness.py`:
- Around line 23-35: The _load_tool_module function must register the newly
created module in sys.modules before executing it to support intra-module
imports, pickling/dataclass lookups, and repeated reloads: add import sys to the
file and set sys.modules[name] = mod (using the same name passed to
_load_tool_module) immediately after creating mod (before calling
spec.loader.exec_module(mod)); keep the existing spec/loader checks and
exec_module call but ensure the module is inserted into sys.modules first.
In `@tools/domain_extractor.py`:
- Around line 36-39: The try/except around path.read_text only handles
UnicodeDecodeError and can let OSError (e.g., permission errors, broken
symlinks) abort the scan; update the exception handling in the try block that
calls path.read_text to also catch OSError (or a tuple like (UnicodeDecodeError,
OSError)) and continue on those failures so recursive scans skip unreadable
files instead of crashing.
In `@tools/l9_meta_injector.py`:
- Around line 67-70: The loop uses a nested conditional to increment updated:
replace `if _should_process(path): updated += 1 if inject(path) else 0` with a
clearer idiom—either skip early with `if not _should_process(path): continue`
and then `if inject(path): updated += 1`, or compute updated with a generator
expression (e.g., sum(1 for p in root.rglob("*") if _should_process(p) and
inject(p))); reference the loop over root.rglob("*"), the _should_process(path)
predicate, and the inject(path) call to locate and apply the change.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 3fedfe05-7232-4024-aca7-3960c2726391
📒 Files selected for processing (28)
.pre-commit-config.yamlengine/arbitration/engine.pyengine/arbitration/schema.pyengine/packet/bridge.pyl9_core/models.pyreports/GMP-Report-131-Fix-DomainPackLoader-Import-And-Constructor.mdtests/conftest.pytests/integration/test_handlers.pytests/security/test_compliance_security.pytests/security/test_injection.pytests/unit/test_arbitration.pytests/unit/test_compliance_checker.pytests/unit/test_domain_loader.pytests/unit/test_gate_compiler.pytests/unit/test_loader.pytests/unit/test_outcomes.pytests/unit/test_packet_bridge.pytests/unit/test_parameter_resolver.pytests/unit/test_scoring_assembler.pytests/unit/test_sync_and_traversal.pytests/unit/test_sync_generator.pytests/unit/test_traversal_assembler.pytests/unit/test_wave6_dormant_features.pytools/audit_harness.pytools/domain_extractor.pytools/l9_meta_injector.pytools/l9_template_manifest.yamltools/validate_domain.py
| for pattern in "import.*PacketEnvelope|PacketEnvelope\\(" "\\bprint\\(" "\\bOptional\\[" "\\bList\\[" "\\bDict\\["; do | ||
| if git diff --cached --name-only | grep "\.py$" | grep -E "^(engine|tests)/" | | ||
| xargs grep -l -P "$pattern" 2>/dev/null | grep -v __pycache__ | grep -q .; then |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find all current PacketEnvelope import sites and constructor calls in engine/ and tests/
# that the new pre-commit hook would now flag.
rg -nP --type=py -g '!**/__pycache__/**' '^(engine|tests)/' --files-with-matches 2>/dev/null
echo "=== import sites ==="
rg -nP --type=py 'import.*PacketEnvelope' engine tests 2>/dev/null
echo "=== constructor calls ==="
rg -nP --type=py 'PacketEnvelope\(' engine tests 2>/dev/null
echo "=== occurrences inside comments/docstrings (potential false positives) ==="
rg -nP --type=py '^\s*#.*import.*PacketEnvelope|"""[^"]*import.*PacketEnvelope' engine tests 2>/dev/nullRepository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 1528
🏁 Script executed:
# Examine the full pre-commit hook context around lines 117-119
sed -n '110,130p' .pre-commit-config.yamlRepository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 954
🏁 Script executed:
# Check the contract test file to understand why PacketEnvelope imports are needed
head -220 tests/contracts/test_contracts.py | tail -50Repository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 1866
🏁 Script executed:
# Verify if there's a TransportPacket class that tests should migrate to
rg -n 'class TransportPacket' engine tests --type pyRepository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 60
🏁 Script executed:
# Check what other patterns are in the hook and how the engine/packet/ module itself handles this
rg -n 'engine/packet' .pre-commit-config.yamlRepository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 60
🏁 Script executed:
# Check if there's any design doc or migration plan for PacketEnvelope
find . -type f \( -name "*.md" -o -name "*.txt" \) | xargs grep -l "PacketEnvelope\|TransportPacket" 2>/dev/null | head -10Repository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 319
🏁 Script executed:
# Check if engine/packet/packet_store.py and __init__.py actually need the imports or if they're exported
head -50 engine/packet/__init__.pyRepository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 501
🏁 Script executed:
# Verify the packet_envelope.py definition itself
head -200 engine/packet/packet_envelope.py | tail -50Repository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 1563
Fix hook to exempt definition, export, and contract test sites from PacketEnvelope pattern check.
The hook will block legitimate code:
engine/packet/packet_envelope.pydefines PacketEnvelope (constructor calls on lines 271, 379 flagged)engine/packet/__init__.pyexports it publicly (line 16 flagged)tests/contracts/test_contracts.pyimports it intentionally for contract assertions (lines 187, 205 flagged)
The hook should only prevent redefining PacketEnvelope in OTHER files, not block its definition, export, and contract test sites. Add exclusions: --exclude-dir=engine/packet --exclude-dir=tests/contracts or use an exemption list in the hook logic. No false positives were found in comments/docstrings, so the grep-based approach is acceptable if scoped correctly.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In @.pre-commit-config.yaml around lines 117 - 119, The grep hook that scans for
"PacketEnvelope" and other patterns should be changed so it does not flag the
module that defines/exports PacketEnvelope or the contract test files; modify
the loop that iterates patterns (the for pattern in
"import.*PacketEnvelope|PacketEnvelope\\(" ... ) to skip the package that
defines PacketEnvelope and the contract-tests directory by adding exclusions
(e.g., add --exclude-dir for the packet package and contract tests or maintain
an explicit exemption list of filenames like the defining module, its __init__
export, and the contract test files) so only other files are prevented from
redefining/using PacketEnvelope. Ensure the adjusted git diff/grep pipeline
still searches the intended engine/tests scope but filters out those exempted
locations.
| def inflate_ingress( | ||
| self, *, tenant_id: str, actor: str, packet_type: str, payload: dict[str, Any] | ||
| ) -> TransportPacket: | ||
| enriched = {**payload, "tenant_id": tenant_id, "actor": actor} | ||
| return TransportPacket.create(action=packet_type, payload=enriched) |
There was a problem hiding this comment.
Mixing routing/identity fields into payload loses structural separation.
inflate_ingress and attach_entity_semantics flatten tenant_id, actor, entity_type, and canonical_entity_type into the payload dict. Two concerns:
- Caller key collision. A caller that passes
payload={"tenant_id": "...", "entity_type": "..."}will have its values silently overwritten — these are last-write-wins via{**payload, "tenant_id": tenant_id, ...}. Worth either prefixing (_meta.tenant_id) or pulling these onto first-class fields ofTransportPacket. - Schema erosion. Downstream consumers previously had a typed
tenant_idon the envelope; now they must reach into adict[str, Any]payload, which defeats type-checking and makes tenant scoping easier to forget. With Neo4j queries required to "scope to tenant database", losing this typed surface is a regression worth tracking.
This may be intentional during cutover, but it's a meaningful API change worth calling out.
Also applies to: 15-23
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@engine/packet/bridge.py` around lines 9 - 13, inflate_ingress (and
attach_entity_semantics) currently flattens routing/identity keys into the
payload which overwrites caller keys and erodes the typed envelope; change these
functions to stop merging tenant_id/actor/entity_type/canonical_entity_type into
payload and instead set them on TransportPacket's dedicated envelope/metadata
fields (e.g., TransportPacket.create(..., metadata={...}) or explicit
TransportPacket.tenant_id/actor fields), or alternatively prefix them (e.g.,
_meta.tenant_id) if you cannot change the TransportPacket shape; ensure you do
not perform {**payload, "tenant_id": ...} merging so caller keys aren't
overwritten and the envelope retains typed routing fields.
| def _stable_hash(payload: dict[str, Any]) -> str: | ||
| canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) | ||
| return sha256(canonical.encode("utf-8")).hexdigest() | ||
|
|
||
| class TenantContext(BaseModel): | ||
| model_config = ConfigDict(frozen=True) | ||
|
|
||
| tenant_id: str | ||
| actor: str | ||
| @dataclass(frozen=True) | ||
| class TransportPacket: | ||
| packet_id: str | ||
| action: str | ||
| payload: dict[str, Any] | ||
| runtime_authority: str = "Gate_SDK" | ||
| routing_authority: str = "Gate" | ||
| transport_authority: str = "TransportPacket" | ||
| content_hash: str = field(init=False) | ||
|
|
||
| def __post_init__(self) -> None: | ||
| object.__setattr__( | ||
| self, | ||
| "content_hash", | ||
| _stable_hash( | ||
| { | ||
| "packet_id": self.packet_id, | ||
| "action": self.action, | ||
| "payload": self.payload, | ||
| "runtime_authority": self.runtime_authority, | ||
| "routing_authority": self.routing_authority, | ||
| "transport_authority": self.transport_authority, | ||
| } | ||
| ), | ||
| ) |
There was a problem hiding this comment.
content_hash includes packet_id, defeating content-addressed semantics.
_stable_hash is computed over a dict that includes packet_id, and TransportPacket.create() assigns a fresh uuid4() every call. As a result:
- Two packets with identical
(action, payload, …authority)tuples produce differentcontent_hashvalues, so a SHA-256 UNIQUE DB constraint oncontent_hashprovides no deduplication — every row is unique by construction. - The companion test
test_transport_packet_hash_changes_with_payload(intests/unit/test_packet_bridge.py) becomes tautologically true: it would pass even if the payload were unchanged, because the randompacket_idalone changes the hash.
This deviates from the documented contract that "Content hash (SHA-256) is a UNIQUE DB constraint" — that constraint is only meaningful if the hash is deterministic over content.
🐛 Proposed fix — exclude `packet_id` from the content hash
def __post_init__(self) -> None:
object.__setattr__(
self,
"content_hash",
_stable_hash(
{
- "packet_id": self.packet_id,
"action": self.action,
"payload": self.payload,
"runtime_authority": self.runtime_authority,
"routing_authority": self.routing_authority,
"transport_authority": self.transport_authority,
}
),
)Based on learnings: "PacketEnvelope is frozen (Pydantic frozen=True). Mutations create new packets via .derive(). Content hash (SHA-256) is a UNIQUE DB constraint."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@l9_core/models.py` around lines 16 - 45, The content hash currently includes
packet_id, which makes content-addressing ineffective; update TransportPacket so
content_hash excludes packet_id by computing _stable_hash over only the actual
content fields (action, payload, runtime_authority, routing_authority,
transport_authority) instead of including packet_id; modify
TransportPacket.__post_init__ to call _stable_hash with a dict that omits
"packet_id" (and ensure any helper like TransportPacket.create or .derive
continues to set packet_id separately), keeping the field name content_hash and
using the existing _stable_hash function.
| def derive(self, action: str | None = None, payload: dict[str, Any] | None = None) -> TransportPacket: | ||
| next_action = action if action is not None else self.action | ||
| next_payload = payload if payload is not None else dict(self.payload) | ||
| return TransportPacket.create(action=next_action, payload=next_payload) |
There was a problem hiding this comment.
derive() drops lineage — violates packet-protocol contract.
The new derive() simply forwards to TransportPacket.create(...), which mints a fresh packet_id and stores nothing about the source packet. There is no parent_id, root_id, generation counter, hop_trace, or delegation_chain. After cutover, every "derived" packet will be indistinguishable from a brand-new root, and provenance / hop-tracing across bridge.py's attach_entity_semantics / decision_packet / outcome_packet chain is lost.
Either the canonical model needs first-class lineage fields (and derive() must populate them), or the lineage contract documented for the packet layer needs to be explicitly retired and the corresponding governance/audit code paths updated.
Based on learnings: "Every derived packet sets parent_id, root_id, increments generation. Hop_trace is append-only with HopEntry. Delegation_chain carries scoped authorization."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@l9_core/models.py` around lines 51 - 54, The derive method currently mints a
fresh TransportPacket and drops provenance; update derive (in TransportPacket /
models.py) to preserve and extend lineage instead of calling
TransportPacket.create(...) with only action/payload: set
parent_id=self.packet_id, set root_id=self.root_id or self.packet_id if root_id
is None, set generation=(self.generation or 0)+1, copy
delegation_chain=self.delegation_chain, and set hop_trace=self.hop_trace + [new
HopEntry(...)] (create a HopEntry representing this derivation); then call
TransportPacket.create or its constructor with these explicit lineage fields so
downstream code (bridge.py functions attach_entity_semantics, decision_packet,
outcome_packet) retains parent_id/root_id/generation/hop_trace/delegation_chain.
Ensure field names match the TransportPacket schema and use existing HopEntry
type for the appended entry.
| @dataclass(frozen=True) | ||
| class PacketEnvelope: | ||
| """Deprecated compatibility wrapper. | ||
|
|
||
| class PacketEnvelope(BaseModel): | ||
| model_config = ConfigDict(frozen=True) | ||
| This type must not be used as canonical runtime truth. | ||
| """ | ||
|
|
||
| packet_id: str = Field(default_factory=lambda: str(uuid4())) | ||
| packet_type: str | ||
| tenant: TenantContext | ||
| action: str | ||
| payload: dict[str, Any] | ||
| lineage: PacketLineage | ||
| created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) | ||
| content_hash: str = "" | ||
|
|
||
| @field_validator("packet_type") | ||
| @classmethod | ||
| def validate_packet_type(cls, value: str) -> str: | ||
| if value != value.lower() or "-" in value or " " in value: | ||
| raise ValueError("packet_type must be lowercase snake_case") | ||
| return value | ||
|
|
||
| def model_post_init(self, __context: Any) -> None: | ||
| if not self.content_hash: | ||
| object.__setattr__(self, "content_hash", self.compute_content_hash()) | ||
|
|
||
| def compute_content_hash(self) -> str: | ||
| canonical_json = json.dumps( | ||
| { | ||
| "packet_type": self.packet_type, | ||
| "tenant": self.tenant.model_dump(mode="json"), | ||
| "payload": self.payload, | ||
| "lineage": self.lineage.model_dump(mode="json"), | ||
| }, | ||
| sort_keys=True, | ||
| separators=(",", ":"), | ||
| default=str, | ||
| ) | ||
| return sha256(canonical_json.encode("utf-8")).hexdigest() | ||
|
|
||
| def derive(self, packet_type: str, payload: dict[str, Any]) -> "PacketEnvelope": | ||
| return PacketEnvelope( | ||
| packet_type=packet_type, | ||
| tenant=self.tenant, | ||
| payload=payload, | ||
| lineage=PacketLineage( | ||
| root_id=self.lineage.root_id, | ||
| parent_id=self.packet_id, | ||
| hop_count=self.lineage.hop_count + 1, | ||
| ), | ||
| ) | ||
|
|
||
| def from_transport_packet(cls, packet: TransportPacket) -> PacketEnvelope: | ||
| return cls(packet_type="compatibility_only", action=packet.action, payload=dict(packet.payload)) | ||
|
|
||
| def make_root_packet(packet_type: str, tenant_id: str, actor: str, payload: dict[str, Any]) -> PacketEnvelope: | ||
| packet_id = str(uuid4()) | ||
| return PacketEnvelope( | ||
| packet_id=packet_id, | ||
| packet_type=packet_type, | ||
| tenant=TenantContext(tenant_id=tenant_id, actor=actor), | ||
| payload=payload, | ||
| lineage=PacketLineage(root_id=packet_id, parent_id=None, hop_count=0), | ||
| ) | ||
| def to_transport_packet(self) -> TransportPacket: | ||
| return TransportPacket.create(action=self.action, payload=dict(self.payload)) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm both PacketEnvelope class definitions coexist on this branch.
ast-grep --pattern 'class PacketEnvelope($$$):
$$$'
# Confirm callers across the repo to gauge import-resolution divergence.
rg -nP '\bfrom\s+(l9_core\.models|engine\.packet(\.\w+)?)\s+import\s+[^#]*\bPacketEnvelope\b'Repository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 9955
🏁 Script executed:
# Search for any imports of PacketEnvelope from l9_core.models specifically
rg -n 'from\s+l9_core\.models\s+import.*PacketEnvelope' --type py
# Search for any usage of l9_core.models at all that might import PacketEnvelope
rg -n 'from\s+l9_core\.models\s+import' --type py
# Check if l9_core.models is imported directly anywhere
rg -n 'import\s+l9_core\.models' --type pyRepository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 312
🏁 Script executed:
# Check for wildcard imports from l9_core.models
rg -n 'from\s+l9_core\.models\s+import\s+\*' --type py
# Look at l9_core/models.py to understand what it exports and why PacketEnvelope is there
head -100 l9_core/models.py
# Check if __all__ is defined in l9_core/models.py
rg -n '__all__' l9_core/models.pyRepository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 2472
🏁 Script executed:
# Check if l9_core/models.py exists in the repo and its status
git ls-files l9_core/models.py
# Check the git log to see recent changes to this file
git log --oneline -5 l9_core/models.py 2>/dev/null || echo "File exists but no recent changes"
# Get file info
stat l9_core/models.py 2>/dev/null | head -5Repository: cryptoxdog/Cognitive.Engine.Graphs
Length of output: 170
Remove or consolidate PacketEnvelope definitions to resolve SHARED-001 contract violation.
Two PacketEnvelope class definitions coexist:
l9_core/models.py:57–73(deprecated dataclass wrapper, unused)engine/packet/packet_envelope.py:173(canonical Pydantic model, actively used)
Per coding guidelines, this violates SHARED-001: "Redefining class PacketEnvelope → contract 7 violation. Use shared model from engine.packet. Merge blocked."
Although l9_core.models.PacketEnvelope is marked deprecated and not imported elsewhere, the dual definition breaches the contract. Either delete the l9_core version entirely (since the canonical form in engine.packet is the source of truth), or move the compatibility shim into engine.packet and re-export from there for backward compatibility during the cutover period. The .pre-commit-config.yaml terminology guard won't detect this collision since both are class declarations.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@l9_core/models.py` around lines 57 - 73, Remove the duplicate dataclass
PacketEnvelope in l9_core.models: eliminate the deprecated PacketEnvelope class
(including its from_transport_packet and to_transport_packet methods) so the
canonical Pydantic PacketEnvelope in engine.packet.packet_envelope.py remains
the single source of truth; alternatively, if a compatibility shim is required
during cutover, move the conversion helpers (from_transport_packet,
to_transport_packet) into engine.packet.packet_envelope.py or add a thin wrapper
there and re-export it for backward compatibility, ensuring all imports
reference engine.packet.packet_envelope.PacketEnvelope and no class definition
remains in l9_core.models.
| TEXT_EXTENSIONS = {".py", ".md", ".yaml", ".yml", ".txt", ".json"} | ||
|
|
||
| def format_comment_block(meta: FileMeta, engine: str, prefix: str = "# ") -> str: | ||
| """Format L9_META as comment block for YAML, shell, Makefile, Dockerfile, etc.""" | ||
| lines = [ | ||
| f"{prefix}", | ||
| ] | ||
| return "\n".join(lines) | ||
|
|
||
|
|
||
| def format_html_comment(meta: FileMeta, engine: str) -> str: | ||
| """Format L9_META as HTML comment for Markdown files.""" | ||
| lines = [ | ||
| "<!-- L9_META", | ||
| f"l9_schema: {L9_SCHEMA_VERSION}", | ||
| f"origin: {meta.origin}", | ||
| f"engine: {engine}", | ||
| f"layer: {_yaml_list(meta.layer)}", | ||
| f"tags: {_yaml_list(meta.tags)}", | ||
| f"owner: {meta.owner}", | ||
| f"status: {meta.status}", | ||
| "/L9_META -->", | ||
| ] | ||
| return "\n".join(lines) | ||
|
|
||
|
|
||
| def format_python_docstring_block(meta: FileMeta, engine: str) -> str: | ||
| """Format L9_META for Python docstring insertion.""" | ||
| lines = [ | ||
| "", | ||
| ] | ||
| return "\n".join(lines) | ||
|
|
||
|
|
||
| def format_json_meta(meta: FileMeta, engine: str) -> dict[str, Any]: | ||
| return { | ||
| "l9_schema": L9_SCHEMA_VERSION, | ||
| "origin": meta.origin, | ||
| "engine": engine, | ||
| "layer": meta.layer, | ||
| "tags": meta.tags, | ||
| "owner": meta.owner, | ||
| "status": meta.status, | ||
| } | ||
|
|
||
|
|
||
| def format_toml_block(meta: FileMeta, engine: str) -> str: | ||
| layer_str = ", ".join(f'"{l}"' for l in meta.layer) | ||
| tags_str = ", ".join(f'"{t}"' for t in meta.tags) | ||
| lines = [ | ||
| "", | ||
| "[tool.l9_meta]", | ||
| f"l9_schema = {L9_SCHEMA_VERSION}", | ||
| f'origin = "{meta.origin}"', | ||
| f'engine = "{engine}"', | ||
| f"layer = [{layer_str}]", | ||
| f"tags = [{tags_str}]", | ||
| f'owner = "{meta.owner}"', | ||
| f'status = "{meta.status}"', | ||
| ] | ||
| return "\n".join(lines) | ||
|
|
||
|
|
||
| # ============================================================================= | ||
| # STRIP EXISTING — Regex patterns for idempotent re-injection | ||
| # ============================================================================= | ||
|
|
||
| # Matches comment-style L9_META blocks (YAML, shell, Makefile, Dockerfile, Python without docstring) | ||
| # Example: | ||
| RE_COMMENT_META = re.compile( | ||
| r"^[ \t]*#[ \t]*---[ \t]*L9_META[ \t]*---.*?#[ \t]*---[ \t]*/L9_META[ \t]*---[ \t]*\n?", | ||
| re.MULTILINE | re.DOTALL, | ||
| ) | ||
|
|
||
| # Matches HTML comment L9_META blocks (Markdown) | ||
| # Example: | ||
| # <!-- L9_META | ||
| # l9_schema: 1 | ||
| # ... | ||
| # /L9_META --> | ||
| RE_HTML_META = re.compile( | ||
| r"<!-- L9_META.*?/L9_META -->[ \t]*\n?", | ||
| re.DOTALL, | ||
| ) | ||
| HEADER_TEMPLATE = """# L9_META: | ||
| # runtime_authority: Gate_SDK | ||
| # routing_authority: Gate | ||
| # canonical_transport: TransportPacket | ||
| # packet_envelope_status: deprecated_compatibility_only | ||
| # cutover_mode: strict | ||
| """ |
There was a problem hiding this comment.
.json (and .md) cannot accept a #-prefixed header — this will corrupt files.
TEXT_EXTENSIONS includes .json, but JSON has no comment syntax. Prepending # L9_META: lines to any .json file under root produces invalid JSON, which will break every consumer that parses these files (configs, lockfiles, fixtures, manifests including tools/l9_template_manifest.yaml siblings, etc.). Markdown is also affected: a leading # L9_META: is parsed as an H1 heading and clobbers doc rendering / front-matter conventions used elsewhere in the repo.
The previous per-extension registry-driven injector handled this by emitting language-appropriate comment delimiters (e.g., wrapping JSON metadata in a sidecar or skipping it). The simplified cutover dropped that without a replacement.
At minimum, drop .json from TEXT_EXTENSIONS and either drop .md or emit an HTML-comment block (<!-- L9_META: ... -->) for it.
🛠 Suggested minimum fix (drop unsafe extensions)
-TEXT_EXTENSIONS = {".py", ".md", ".yaml", ".yml", ".txt", ".json"}
+# Only file types whose first-column "#" is a valid line comment.
+TEXT_EXTENSIONS = {".py", ".yaml", ".yml", ".txt"}If .md injection is required, branch HEADER_TEMPLATE per-extension and emit an HTML comment for Markdown; for .json use a sidecar (e.g., foo.json.l9meta) rather than mutating the file.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| TEXT_EXTENSIONS = {".py", ".md", ".yaml", ".yml", ".txt", ".json"} | |
| def format_comment_block(meta: FileMeta, engine: str, prefix: str = "# ") -> str: | |
| """Format L9_META as comment block for YAML, shell, Makefile, Dockerfile, etc.""" | |
| lines = [ | |
| f"{prefix}", | |
| ] | |
| return "\n".join(lines) | |
| def format_html_comment(meta: FileMeta, engine: str) -> str: | |
| """Format L9_META as HTML comment for Markdown files.""" | |
| lines = [ | |
| "<!-- L9_META", | |
| f"l9_schema: {L9_SCHEMA_VERSION}", | |
| f"origin: {meta.origin}", | |
| f"engine: {engine}", | |
| f"layer: {_yaml_list(meta.layer)}", | |
| f"tags: {_yaml_list(meta.tags)}", | |
| f"owner: {meta.owner}", | |
| f"status: {meta.status}", | |
| "/L9_META -->", | |
| ] | |
| return "\n".join(lines) | |
| def format_python_docstring_block(meta: FileMeta, engine: str) -> str: | |
| """Format L9_META for Python docstring insertion.""" | |
| lines = [ | |
| "", | |
| ] | |
| return "\n".join(lines) | |
| def format_json_meta(meta: FileMeta, engine: str) -> dict[str, Any]: | |
| return { | |
| "l9_schema": L9_SCHEMA_VERSION, | |
| "origin": meta.origin, | |
| "engine": engine, | |
| "layer": meta.layer, | |
| "tags": meta.tags, | |
| "owner": meta.owner, | |
| "status": meta.status, | |
| } | |
| def format_toml_block(meta: FileMeta, engine: str) -> str: | |
| layer_str = ", ".join(f'"{l}"' for l in meta.layer) | |
| tags_str = ", ".join(f'"{t}"' for t in meta.tags) | |
| lines = [ | |
| "", | |
| "[tool.l9_meta]", | |
| f"l9_schema = {L9_SCHEMA_VERSION}", | |
| f'origin = "{meta.origin}"', | |
| f'engine = "{engine}"', | |
| f"layer = [{layer_str}]", | |
| f"tags = [{tags_str}]", | |
| f'owner = "{meta.owner}"', | |
| f'status = "{meta.status}"', | |
| ] | |
| return "\n".join(lines) | |
| # ============================================================================= | |
| # STRIP EXISTING — Regex patterns for idempotent re-injection | |
| # ============================================================================= | |
| # Matches comment-style L9_META blocks (YAML, shell, Makefile, Dockerfile, Python without docstring) | |
| # Example: | |
| RE_COMMENT_META = re.compile( | |
| r"^[ \t]*#[ \t]*---[ \t]*L9_META[ \t]*---.*?#[ \t]*---[ \t]*/L9_META[ \t]*---[ \t]*\n?", | |
| re.MULTILINE | re.DOTALL, | |
| ) | |
| # Matches HTML comment L9_META blocks (Markdown) | |
| # Example: | |
| # <!-- L9_META | |
| # l9_schema: 1 | |
| # ... | |
| # /L9_META --> | |
| RE_HTML_META = re.compile( | |
| r"<!-- L9_META.*?/L9_META -->[ \t]*\n?", | |
| re.DOTALL, | |
| ) | |
| HEADER_TEMPLATE = """# L9_META: | |
| # runtime_authority: Gate_SDK | |
| # routing_authority: Gate | |
| # canonical_transport: TransportPacket | |
| # packet_envelope_status: deprecated_compatibility_only | |
| # cutover_mode: strict | |
| """ | |
| # Only file types whose first-column "#" is a valid line comment. | |
| TEXT_EXTENSIONS = {".py", ".yaml", ".yml", ".txt"} | |
| HEADER_TEMPLATE = """# L9_META: | |
| # runtime_authority: Gate_SDK | |
| # routing_authority: Gate | |
| # canonical_transport: TransportPacket | |
| # packet_envelope_status: deprecated_compatibility_only | |
| # cutover_mode: strict | |
| """ |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tools/l9_meta_injector.py` around lines 13 - 21, TEXT_EXTENSIONS currently
includes ".json" and ".md" which makes HEADER_TEMPLATE (prefixed "#" lines)
unsafe: remove ".json" from TEXT_EXTENSIONS and either remove ".md" or make
injection conditional by branching on extension in the injector (use an HTML
comment wrapper like "<!-- L9_META: ... -->" for ".md"); for ".json" avoid
in-file mutation (emit a sidecar file such as "<original>.l9meta" or
"<original>.json.l9meta") instead. Update the logic that uses TEXT_EXTENSIONS
and HEADER_TEMPLATE so it selects the proper per-extension header or sidecar
behavior rather than blindly prepending HEADER_TEMPLATE to all entries.
| def inject(path: Path) -> bool: | ||
| text = path.read_text(encoding="utf-8") | ||
| if _has_header(text): | ||
| return False | ||
| header = HEADER_TEMPLATE + _compatibility_hint(text) | ||
| path.write_text(header + text, encoding="utf-8") | ||
| return True |
There was a problem hiding this comment.
inject() has no error handling for non-UTF-8 or unreadable files.
read_text(encoding="utf-8") raises UnicodeDecodeError for any binary-ish file that happens to share a tracked extension (e.g., a .txt or .json with non-UTF-8 bytes), and both read_text / write_text raise OSError on permission/symlink issues. Today a single offending file aborts the whole traversal in main() with a stack trace and a partial run. Catch explicitly and skip with a warning so injection is resumable.
♻️ Suggested fix
def inject(path: Path) -> bool:
- text = path.read_text(encoding="utf-8")
+ try:
+ text = path.read_text(encoding="utf-8")
+ except (UnicodeDecodeError, OSError) as exc:
+ print(f"skip(read): {path}: {exc}")
+ return False
if _has_header(text):
return False
header = HEADER_TEMPLATE + _compatibility_hint(text)
- path.write_text(header + text, encoding="utf-8")
- return True
+ try:
+ path.write_text(header + text, encoding="utf-8")
+ except OSError as exc:
+ print(f"skip(write): {path}: {exc}")
+ return False
+ return True🧰 Tools
🪛 GitHub Check: SonarCloud Code Analysis
[failure] 57-57: Change this code to not construct the path from user-controlled data.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tools/l9_meta_injector.py` around lines 52 - 58, The inject function
currently can raise UnicodeDecodeError or OSError and abort the run; wrap the
file operations in a try/except around path.read_text and path.write_text inside
inject (while keeping the existing _has_header, HEADER_TEMPLATE and
_compatibility_hint usage) and catch UnicodeDecodeError and OSError, log a
warning (include path and exception) and return False to skip the file so
traversal in main can continue; ensure you only swallow these expected file I/O
errors and let unexpected exceptions propagate.
| def main() -> int: | ||
| parser = argparse.ArgumentParser(description="Inject strict cutover metadata") | ||
| parser.add_argument("root", type=Path) | ||
| args = parser.parse_args() | ||
|
|
||
| root = Path(args.root).resolve() | ||
| engine = args.engine | ||
| updated = skipped = missing = errors = 0 | ||
|
|
||
| for fm in FILE_REGISTRY: | ||
| fpath = root / fm.path | ||
| if not fpath.exists(): | ||
| print(f" MISSING {fm.path}") | ||
| missing += 1 | ||
| continue | ||
| try: | ||
| content = fpath.read_text(encoding="utf-8") | ||
| new_content = inject_meta(content, fm, engine) | ||
| if new_content == content: | ||
| print(f" SKIP {fm.path} (already has correct meta)") | ||
| skipped += 1 | ||
| continue | ||
| if args.apply: | ||
| fpath.write_text(new_content, encoding="utf-8") | ||
| print(f" UPDATED {fm.path}") | ||
| else: | ||
| print(f" PENDING {fm.path} (dry-run)") | ||
| updated += 1 | ||
| except Exception as e: | ||
| print(f" ERROR {fm.path} ({e})") | ||
| errors += 1 | ||
| root = args.root.resolve() | ||
| updated = 0 | ||
| for path in root.rglob("*"): | ||
| if _should_process(path): | ||
| updated += 1 if inject(path) else 0 | ||
|
|
||
| print(f"\n{'=' * 60}") | ||
| print(f" Engine: {engine}") | ||
| print(f" Updated: {updated}") | ||
| print(f" Skipped: {skipped}") | ||
| print(f" Missing: {missing}") | ||
| print(f" Errors: {errors}") | ||
| print(f" Total: {len(FILE_REGISTRY)}") | ||
| if not args.apply and updated > 0: | ||
| print("\n >>> Dry-run mode. Run with --apply to write changes.") | ||
| print(f"{'=' * 60}") | ||
| print(f"updated_files={updated}") | ||
| return 0 |
There was a problem hiding this comment.
main() has no exclusion list and no dry-run/apply guard — destructive blast radius.
root.rglob("*") will descend into .git/, .venv/, node_modules/, build/, dist/, vendored third-party trees, generated artifacts, and test fixtures. Combined with the unconditional path.write_text(...) in inject() (line 57) and the removal of the prior --apply/dry-run flow, a single accidental run can rewrite hundreds of files outside the engine source tree. The summary count printed on line 72 (updated_files=…) gives no path-level visibility to recover from a bad invocation.
Recommend (a) excluding well-known non-source dirs, (b) restoring a default-dry-run with explicit --apply, and (c) listing changed paths.
🛡 Suggested guardrails
+EXCLUDE_DIR_NAMES = {
+ ".git", ".venv", "venv", "env", "node_modules",
+ "build", "dist", ".tox", ".mypy_cache", ".ruff_cache",
+ "__pycache__", ".pytest_cache", "site-packages",
+}
+
+
+def _is_excluded(path: Path, root: Path) -> bool:
+ return any(part in EXCLUDE_DIR_NAMES for part in path.relative_to(root).parts)
+
+
def main() -> int:
parser = argparse.ArgumentParser(description="Inject strict cutover metadata")
parser.add_argument("root", type=Path)
+ parser.add_argument("--apply", action="store_true",
+ help="Actually write changes (default: dry-run).")
args = parser.parse_args()
root = args.root.resolve()
updated = 0
+ changed: list[Path] = []
for path in root.rglob("*"):
- if _should_process(path):
- updated += 1 if inject(path) else 0
+ if not _should_process(path) or _is_excluded(path, root):
+ continue
+ if not args.apply:
+ # Dry-run: only report would-update files.
+ text = path.read_text(encoding="utf-8", errors="replace")
+ if not _has_header(text):
+ changed.append(path)
+ continue
+ if inject(path):
+ updated += 1
+ changed.append(path)
- print(f"updated_files={updated}")
+ for p in changed:
+ print(p)
+ print(f"updated_files={updated} dry_run={not args.apply}")
return 0🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tools/l9_meta_injector.py` around lines 61 - 73, main() currently walks
root.rglob("*") with no exclusions and always calls inject(path) which uses
path.write_text, risking mass destructive edits; update main() to (1) add an
explicit --apply boolean flag defaulting to False so runs are dry-run by default
and only write when --apply is provided, (2) build an exclusion set of
well-known non-source dirs (e.g., ".git", ".venv", "node_modules", "build",
"dist", vendor/generated/test fixture dirs) and skip any paths under those in
the walk (use _should_process(path) or enhance it to check the exclusion set),
and (3) collect and print the list of changed file paths (instead of just a
count) when dry-run or after applying so users can review edits; ensure
inject(path) only calls path.write_text when the --apply flag is True.
| forbidden_primary_signals: | ||
| - "PacketEnvelope" | ||
| - "inflate_ingress(" | ||
| - "deflate_egress(" | ||
| - "chassis.router" | ||
| strict_fail_closed: true |
There was a problem hiding this comment.
Manifest and tool token sets are out of sync.
forbidden_primary_signals lists four tokens, but enforcement diverges:
tools/domain_extractor.pySIGNALS["deprecated_compatibility"]is missing"chassis.router".tools/validate_domain.pyFORBIDDEN_VALUESuses"inflate_ingress"/"deflate_egress"(no trailing"("), which matches benign identifiers likeinflate_ingress_metricsand is also missing"chassis.router".
With strict_fail_closed: true, these tools should be the single source of truth's enforcement arm. Suggest having both tools load this manifest (e.g., yaml.safe_load(Path("tools/l9_template_manifest.yaml").read_text())) and derive their token sets from forbidden_primary_signals/required_signals/authorities so the manifest is the only place tokens are declared.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tools/l9_template_manifest.yaml` around lines 31 - 36, The manifest's token
lists (forbidden_primary_signals / required_signals / authorities) must be the
single source of truth; update tools/domain_extractor.py
(SIGNALS["deprecated_compatibility"]) and tools/validate_domain.py
(FORBIDDEN_VALUES) to load and derive their token sets from the YAML manifest
(e.g., yaml.safe_load(Path("tools/l9_template_manifest.yaml").read_text()))
instead of hardcoding values so entries like "chassis.router" and the exact
forbidden tokens (including parentheses where present) stay in sync with
strict_fail_closed enforcement. Ensure both modules map the manifest keys to
their internal structures (SIGNALS[...] and FORBIDDEN_VALUES) and remove the
duplicated hardcoded lists.
| FORBIDDEN_VALUES = { | ||
| "PacketEnvelope", | ||
| "inflate_ingress", | ||
| "deflate_egress", | ||
| } | ||
| REQUIRED_AUTHORITY = { | ||
| "runtime_authority": "Gate_SDK", | ||
| "routing_authority": "Gate", | ||
| "canonical_transport": "TransportPacket", | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Hardcoded rules drift from l9_template_manifest.yaml and ignore allowed_only_in contexts.
Two issues:
FORBIDDEN_VALUESandREQUIRED_AUTHORITYduplicate values declared intools/l9_template_manifest.yaml(forbidden_primary_signals,authorities). They will silently drift; e.g., the manifest already listschassis.routeras forbidden but it is absent here. Load the manifest once and derive these constants from it.FORBIDDEN_VALUESuses bare strings ("inflate_ingress"without trailing"("), which matches benign identifiers likeinflate_ingress_count. The manifest uses"inflate_ingress("/"deflate_egress("— keep parity.- The manifest declares
deprecated_compatibility.packet_envelope.allowed_only_in: [compatibility_shims, migration_docs, compatibility_tests, scanner_allowlists], butvalidate_fileflags every match unconditionally. Migration/compat YAML will produce CRITICAL false positives, which will be amplified bystrict_fail_closed: true.
♻️ Sketch: load from manifest + honor allowed contexts
-FORBIDDEN_VALUES = {
- "PacketEnvelope",
- "inflate_ingress",
- "deflate_egress",
-}
-REQUIRED_AUTHORITY = {
- "runtime_authority": "Gate_SDK",
- "routing_authority": "Gate",
- "canonical_transport": "TransportPacket",
-}
+_MANIFEST = yaml.safe_load(
+ (Path(__file__).parent / "l9_template_manifest.yaml").read_text(encoding="utf-8")
+)
+FORBIDDEN_VALUES: set[str] = set(_MANIFEST.get("forbidden_primary_signals", []))
+REQUIRED_AUTHORITY: dict[str, str] = {
+ "runtime_authority": _MANIFEST["authorities"]["runtime"],
+ "routing_authority": _MANIFEST["authorities"]["routing"],
+ "canonical_transport": _MANIFEST["authorities"]["transport"],
+}
+ALLOWED_CONTEXTS: set[str] = set(
+ _MANIFEST.get("deprecated_compatibility", {})
+ .get("packet_envelope", {})
+ .get("allowed_only_in", [])
+)Then in validate_file, skip the forbidden-token check when path lives under one of the allowed context directories.
Also applies to: 62-71
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tools/validate_domain.py` around lines 18 - 27, Replace the hardcoded
FORBIDDEN_VALUES and REQUIRED_AUTHORITY with values loaded at runtime from
tools/l9_template_manifest.yaml (derive forbidden_primary_signals and
authorities once at module init), update forbidden tokens to match the manifest
exactly (e.g., use "inflate_ingress(" and "deflate_egress(" rather than bare
identifiers) and update any occurrences at lines referenced; then modify
validate_file to consult the manifest's
deprecated_compatibility.packet_envelope.allowed_only_in (or equivalent
allowed_only_in entries) and skip the forbidden-token check when the file path
lives under any allowed context directories (e.g., compatibility_shims,
migration_docs, compatibility_tests, scanner_allowlists), ensuring behavior
mirrors the manifest-derived rules rather than unconditional flagging.
|
2 participants
Summary
l9_core/models.py(frozen dataclass with SHA-256 content hashing).PacketEnvelopeis retained as a deprecated compatibility wrapper withfrom_transport_packet()/to_transport_packet()bridge methods.PacketEnvelope(imports + constructor calls) while ignoring deprecation documentation — zero false positives on files that merely reference PacketEnvelope's deprecated status.audit_harness.py,validate_domain.py,domain_extractor.py,l9_meta_injector.py) refactored from subprocess toimportlib.utildirect module loading — eliminates S603 security warnings entirely.Changes
Core Architecture
l9_core/models.pyTransportPacket(canonical) +PacketEnvelope(deprecated shim)engine/packet/bridge.pyTransportPacketAPIengine/arbitration/schema.pyDecisionPolicy,HardConstraint,PolicyWeights,PolicyThresholdsengine/arbitration/engine.pyCI Enforcement
.pre-commit-config.yamlterminology-guardregex:import.*PacketEnvelope|PacketEnvelope\\(— targets active usage onlyTools (Pack 3)
tools/audit_harness.pytools/validate_domain.pytools/domain_extractor.pytools/l9_meta_injector.pytools/l9_template_manifest.yamlPre-existing Fixes (broken on main)
DomainSpecLoader→DomainPackLoaderimport across 14 test filesDomainPackLoaderconstructor:domains_dir→config_pathtests/conftest.py: added missingqueryentityfield to minimal spec fixturetest_wave6_dormant_features: fixed LLM testmatchregex ("OPENAI_API_KEY")test_injection/test_sync_generator: fixed RUF059 unused unpacked varsGovernance Compliance
TransportPacketas canonical;PacketEnvelopeflagged for active usage but not for deprecation referencesTransportPacketisfrozen=Truewithcontent_hashcomputed in__post_init__Test Impact
Test plan
Made with Cursor
Summary by CodeRabbit
New Features
Refactor