Repository navigation
Expose durable taxi event-bound repair - #4
Conversation
|
Warning Review limit reached
Next review available in: 37 minutes You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughThe PR adds durable event and observation validation, event-bound repair and publication recovery, a CLI watcher and injector, and expanded integration coverage. Example lineage records now use updated document identifiers and integrity hashes. ChangesEvent-bound repair workflow
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant Operator
participant TaxiCLI
participant MonitoringStore
participant TaxiSlice
participant RepairRegistry
participant PublicationStore
Operator->>TaxiCLI: start watch or inject mutation
TaxiCLI->>MonitoringStore: poll or persist delivered event
TaxiCLI->>TaxiSlice: pass event_id to repair
TaxiSlice->>RepairRegistry: validate binding and repair evidence
TaxiSlice->>PublicationStore: verify publication boundary
PublicationStore-->>TaxiSlice: return verified receipt
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 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 |
|
@coderabbitai resume |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (16)
src/reglineage/cli.py (1)
523-532: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueExtract the fixed mutation ID into a module-level constant.
The literal
"taxi-pulocation-borough-only"appears here and in_watch_taxiat Line 234. The watcher binds the observation to the same mutation that this command creates. A single constant prevents the two sites from drifting.♻️ Proposed refactor
Add the constant near the top of the module:
+_TAXI_MUTATION_ID = "taxi-pulocation-borough-only" + + def _print(value: object) -> None:Then use it at both sites:
elif args.command == "taxi-inject": - mutation_id = "taxi-pulocation-borough-only" + mutation_id = _TAXI_MUTATION_IDservice.mutator.bind_observation( # type: ignore[attr-defined] - "taxi-pulocation-borough-only", event.event_id + _TAXI_MUTATION_ID, event.event_id )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/cli.py` around lines 523 - 532, Define a module-level constant for the fixed “taxi-pulocation-borough-only” mutation ID, then replace the duplicated literal in the taxi-inject command and _watch_taxi with that constant so both paths remain synchronized.tests/integration/test_north_star_benchmark.py (1)
172-180: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSelect the evidence step by its
stepnumber, not by list index.
item["evidence"]["steps"][8]depends on step 9 staying at index 8.scripts/benchmark_north_star.pyLine 827 already maps steps by number for the same evidence structure. Use the same lookup here so an inserted step fails with a clear message instead of asserting against the wrong step.♻️ Proposed refactor
for item in second["repair_repetitions"]: - observed = item["evidence"]["steps"][8] + steps = {int(step["step"]): step for step in item["evidence"]["steps"]} + observed = steps[9] self.assertEqual(observed["observation_truth_label"], "OBSERVED")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_north_star_benchmark.py` around lines 172 - 180, Update the evidence lookup in the repair_repetitions assertions to select the entry whose step number is 9, matching the step-number mapping used by benchmark evidence handling, instead of indexing steps at position 8. Keep the existing assertions unchanged after selecting the correct step.src/reglineage/mcp_backend.py (2)
653-662: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRename the reused
bindinglocal.Line 653 binds
bindingto therepair_event_bindingrecord. Line 673 rebinds the same name to arepair_publication_bindingrecord, and line 677 can set it toNone. The event-binding checks finish before the rebinding, so behavior is correct today. The shared name still makes any future read after line 673 resolve to the wrong record.Use
event_bindingat line 653 andpublication_bindingat line 673.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/mcp_backend.py` around lines 653 - 662, Rename the local loaded with “repair_event_binding” from binding to event_binding and update its proposal_hash and binding_hash checks accordingly; rename the later “repair_publication_binding” local to publication_binding, including its None assignment and subsequent references.
532-539: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winChain the exception cause.
Line 537 raises inside an
except KeyErrorblock withoutfrom. Every other handler in this file usesfrom exc, including_loadat line 159 and_eventat line 171. The missing chain hides the original lookup failure in tracebacks.🐛 Proposed fix to restore the cause chain
try: item = self.service.records.load( "repair_proposal", f"borough-repair:{event.event_id}:v1" ) - except KeyError: - raise ToolDenied("REPAIR_NOT_PROPOSED") + except KeyError as exc: + raise ToolDenied("REPAIR_NOT_PROPOSED") from exc🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/mcp_backend.py` around lines 532 - 539, Update the KeyError handler in the repair-proposal loading flow to catch the exception as a named variable and raise ToolDenied("REPAIR_NOT_PROPOSED") from that original exception, matching the chaining pattern used by _load and _event.Source: Linters/SAST tools
src/reglineage/composition.py (5)
628-655: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSplit the 26-clause predicate into named groups with distinct messages.
Every failure in this condition raises the same message: "Event does not authorize repair of this exact lease". Operators cannot tell whether the observation identity, the policy epoch, the delivery state, or the lease scope caused the rejection. The block also mixes observation checks, event checks, and lease checks in one expression.
Group the clauses into small helpers such as
_observation_binds_event,_event_is_fixed_borough, and_lease_admits_repair, and raise a specific message per group. Behavior stays identical and the code stays fail-closed.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/composition.py` around lines 628 - 655, Refactor the authorization predicate around the repair validation into named helpers such as _observation_binds_event, _event_is_fixed_borough, and _lease_admits_repair, separating observation, event, delivery, and lease checks while preserving every existing condition and fail-closed behavior. Update the validation flow to raise distinct, actionable PermissionError messages for each failed group instead of one generic message.
1273-1275: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTwo files hardcode the repair record id version.
propose_borough_repairbuildsproposal_idasf"borough-repair:{event.event_id}:{REPAIR_RECIPE_VERSION}", but both lookup sites spell the version as the literalv1. IfREPAIR_RECIPE_VERSIONadvances, both lookups raiseKeyErroragainst durable records that exist under the new id.
src/reglineage/composition.py#L1273-L1275: build therepair_event_bindingid withREPAIR_RECIPE_VERSIONinstead of thev1literal.src/reglineage/mcp_backend.py#L533-L535: build therepair_proposalid with the same importedREPAIR_RECIPE_VERSIONinstead of thev1literal.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/composition.py` around lines 1273 - 1275, The repair record lookups must use the shared REPAIR_RECIPE_VERSION instead of hardcoded v1. Update the repair_event_binding lookup in src/reglineage/composition.py:1273-1275 and the repair_proposal lookup in src/reglineage/mcp_backend.py:533-535, ensuring both construct IDs consistently with propose_borough_repair and import the shared version symbol where needed.
862-871: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRead leases through the
CapabilityRuntimeAPI instead of its private table.
_assert_overall_activeopens a secondsqlite3connection toself.runtime.db_pathand selects fromcapability_leases. This duplicates the runtime's schema knowledge insidecomposition.py, bypasses the runtime's own lock, and is the only reasoncomposition.pyimportssqlite3at line 9. A schema change insidereglineage.runtime.storewould break this call site with no type or import signal.Add a narrow accessor to
CapabilityRuntime, for exampleleases_by_tool(tool_id), and call it here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/composition.py` around lines 862 - 871, The _assert_overall_active method should stop querying the private capability_leases table through a separate sqlite3 connection. Add a narrow public leases_by_tool(tool_id) accessor to CapabilityRuntime that performs the runtime-owned lookup, then update _assert_overall_active to use it for OVERALL_TOOL_ID and retain the existing ACTIVE-state validation; remove the now-unused sqlite3 import from composition.py.
858-860: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winName the one-hour approval window.
The one-hour freshness window is a literal here, and
timedelta(hours=1)also appears at line 1023 and line 1115 where approval expiry is set. If one value changes, an approval can be issued with an expiry that the freshness check rejects, or the reverse.Define a module constant such as
_APPROVAL_VALIDITY = timedelta(hours=1)and use it at all three sites.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/composition.py` around lines 858 - 860, Define a module-level _APPROVAL_VALIDITY constant set to the one-hour timedelta, then update _unused_approval_is_current and both approval-expiry assignments to reuse it instead of constructing separate timedelta(hours=1) values.
57-68: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPromote the borrowed
replay.repairhelpers to a public surface.Lines 63-67 import
_candidate_content_hash,_governed_content_hash, and_validation_from_json. The leading underscore marks them private toreglineage.replay.repair, butcomposition.pynow depends on all three in_verify_repair_candidateand_resume_activated_repair. A rename or signature change insidereplay.repairbreakscomposition.pywith no compile-time signal for the signature case.Export these three under public names from
reglineage.replay.repair, or add a small public verification function in that module and call it from here. This keeps the candidate-verification logic on the side that owns the schema.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/composition.py` around lines 57 - 68, Promote the private replay.repair helpers used by _verify_repair_candidate and _resume_activated_repair to a stable public API. Export public equivalents for _candidate_content_hash, _governed_content_hash, and _validation_from_json, or add one public verification function in reglineage.replay.repair and update composition.py to call it, keeping schema-specific candidate verification owned by that module.src/reglineage/monitoring/service.py (1)
423-463: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winDerive the observation stream from the mutation instead of the module constant.
await_observationpollsmutation.previous_snapshot.target; ifinject_borough_onlyaccepts a non-defaulttarget,bind_observationstill readsPULOCATION_TARGET.stream_idandPULOCATION_TARGET.identity. Usemutation.previous_snapshot.targetfor the observation lookup, expectedstream_id, current snapshot lookup, and target-identity comparison so mutation observations bind the actual injected target.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/reglineage/monitoring/service.py` around lines 423 - 463, Update bind_observation to derive the target from mutation.previous_snapshot.target instead of PULOCATION_TARGET. Use that target for observation lookup, expected_event_id stream_id, current snapshot retrieval, and target-identity validation, while preserving the existing mutation binding checks.tests/integration/test_repair_event_contract.py (6)
675-680: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSilence the S608 finding with a scoped suppression.
Ruff reports S608 on this statement. The interpolated value is the module constant
PERSISTED_BOROUGH_RELATION, not caller input, so there is no injection path. A relation name cannot be a bound parameter, so the interpolation must stay. Add a targeted suppression so the lint gate stays clean.🔧 Proposed fix
with duckdb.connect(str(self.service.replay_db)) as connection: - connection.execute( + connection.execute( # noqa: S608 - relation name is a module constant f"UPDATE {PERSISTED_BOROUGH_RELATION} SET borough = 'Tampered' " "WHERE source_record_id = (SELECT MIN(source_record_id) " f"FROM {PERSISTED_BOROUGH_RELATION})" )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_repair_event_contract.py` around lines 675 - 680, Add a narrowly scoped Ruff S608 suppression to the SQL UPDATE statement in the test using PERSISTED_BOROUGH_RELATION, preserving the constant interpolation and query behavior while leaving unrelated SQL checks enabled.Source: Linters/SAST tools
741-746: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReplace the positional index with a named binding.
completed[3]does not state what the fourth element is. A reader cannot tell which lease the assertion checks. Unpack the result, or index by a named constant.♻️ Proposed refactor
completed = self.service.repair( self.zone, event_id=event_id, now=publication_approval.approved_at + timedelta(hours=2), ) - self.assertEqual(completed[3].replacement_for, self.zone.lease_id) + *_, replacement_lease = completed + self.assertEqual(replacement_lease.replacement_for, self.zone.lease_id)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_repair_event_contract.py` around lines 741 - 746, Replace the positional completed[3] access in the repair test with a named binding by unpacking the repair result or using an existing named index constant, then assert replacement_for against self.zone.lease_id through that named reference.
113-118: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDo not re-derive the observation identifier in the test.
The helper reconstructs
observation_idas"poll:" + canonical_sha256(observation_values)[:24]. This copies the production derivation rule, including the prefix and the 24-character truncation. If the production derivation changes, this helper produces a value that no longer matches the real format, and the tampering tests can pass for the wrong reason.Reuse the production helper that builds the observation identifier, and assert the derived value against it.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_repair_event_contract.py` around lines 113 - 118, Update the changed_observation setup around monitoring_hashed to reuse the production helper that derives the observation identifier instead of manually combining "poll:" with canonical_sha256 and truncation. Assert or pass the resulting identifier from that helper while preserving the tampering test’s intended behavior.
72-89: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRemove the unused
deliverparameter, or use it in the undelivered test.No caller passes
deliver=False.test_injected_unobserved_and_observed_undelivered_fail_before_proposalre-implements the undelivered flow inline at lines 195-207. This duplicates the inject-and-poll sequence and lets the two paths drift.Call
self._inject(deliver=False)in that test, or drop the parameter from_inject.Also applies to: 194-213
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_repair_event_contract.py` around lines 72 - 89, Update the undelivered test test_injected_unobserved_and_observed_undelivered_fail_before_proposal to call _inject(deliver=False) instead of duplicating the inject-and-poll flow inline. Preserve the existing assertions and remove the now-redundant setup code; keep _inject’s deliver branching intact.
29-46: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRegister cleanups instead of relying on
tearDown.
setUpcreatesTemporaryDirectoryandTaxiSlicebefore it finishes. Iffixture_cache,TaxiSlice,replay.start, or_leaseraises,tearDowndoes not run and the temporary directory leaks. UseaddCleanupso each resource is released as soon as it is created.Note that
self.serviceis rebound at line 554, so the service cleanup must resolve the current attribute at call time.♻️ Proposed refactor
def setUp(self) -> None: self.temporary = TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) root = Path(self.temporary.name) _, info = fixture_cache(root) self.run_root = root / "run" self.stage_path = str(info["stage_path"]) self.now = datetime.now(timezone.utc) - timedelta(minutes=1) self.service = TaxiSlice( self.run_root, stage_path=self.stage_path, clock_started_at=NOW ) + self.addCleanup(lambda: self.service.close()) self.service.replay.start(batch_size=500_000) self.service.poller.poll() self.zone = self._lease(ZONE_TOOL_ID, ZONE_VERSION, "zone") self.overall = self._lease(OVERALL_TOOL_ID, OVERALL_VERSION, "overall") - def tearDown(self) -> None: - self.service.close() - self.temporary.cleanup() -🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_repair_event_contract.py` around lines 29 - 46, Update the test setup in setUp to register cleanup callbacks immediately after creating TemporaryDirectory and TaxiSlice, so cleanup still runs when later setup steps fail. Use addCleanup for the temporary directory and a callback that resolves self.service at cleanup time, preserving the existing tearDown behavior while handling the later self.service rebinding.
652-657: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse one consistent type for
patch_hash.Line 654 passes
manifest["patch_hash"]directly. Line 689 wraps the same value withstr(...). Pick one form. If the record value is already a string, remove thestr(...)call. If it is not, add the conversion in both places so both tests exercise the same input type.Also applies to: 687-691
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_repair_event_contract.py` around lines 652 - 657, Make the `patch_hash` argument type consistent in both `publication_receipts` assertions: update the call at the shown `publication_manifest` lookup and the corresponding call around the second assertion to either both pass the value directly or both convert it with `str(...)`, based on the record’s actual type.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/reglineage/cli.py`:
- Around line 227-259: Update _watch_taxi to catch KeyError and
MonitoringConflict from service.mutator.bind_observation for
"taxi-pulocation-borough-only". On failure, emit the function’s structured
terminal JSON record and return the distinct failure exit code used by its other
error paths, ensuring no subsequent unguarded processing can crash the watcher;
import MonitoringConflict alongside the existing monitoring symbols.
In `@src/reglineage/composition.py`:
- Around line 702-722: Extract the shared repair-binding lookup and seven-field
validation into an _existing_repair_binding(zone_lease, event, observation)
helper returning the validated binding or None, while preserving both existing
PermissionError messages. In src/reglineage/composition.py lines 702-722,
replace the inline logic with this helper and retain only proposal construction
and persistence; do the same at lines 775-796, retaining only proposal
re-derivation and durable-record comparison.
In `@src/reglineage/rehearsal.py`:
- Around line 354-373: Update the repair call in the rehearsal flow after event
lookup to compute a timestamp later than the rehearsal timestamp, observed event
timestamp, and polled observation timestamp, then pass that clamped value as now
to service.repair. Preserve the existing zone_lease, event_id, and on_suspended
arguments.
In `@src/reglineage/runtime/store.py`:
- Around line 601-606: Update the event lookup flow around
`_validated_event_row` to explicitly check whether `event_row` is missing and
raise `RuntimeContractError` before invoking the helper. Match the existing
missing-event handling used by `accepted_event` and `_status`, while preserving
validation for present rows.
In `@tests/integration/test_repair_event_contract.py`:
- Around line 35-38: Update the test setup around self.now and the later
self.now advances to derive all timestamps from the fixed NOW constant instead
of datetime.now(timezone.utc). Preserve the existing one-minute offset and
subsequent timedelta progression so chronology assertions remain deterministic.
- Around line 450-460: Update the SQLite connection block around the
change_events UPDATE to explicitly close the connection, using
contextlib.closing (and adding its import) or an equivalent explicit close.
Preserve the existing transaction commit behavior and query parameters while
ensuring the handle is released before teardown cleanup.
In `@tests/integration/test_taxi_cli.py`:
- Around line 295-397: Update
test_separate_processes_ready_inject_and_persist_delivered_history to guarantee
cleanup of both watcher and injector processes on every exit, using a test-class
_reap helper or equivalent registered cleanup that kills running children and
waits for them. Replace the unbounded watcher.stdout.readline readiness wait
with a bounded approach, and ensure timeout paths around injector.wait and
watcher.communicate invoke cleanup before propagating the failure.
---
Nitpick comments:
In `@src/reglineage/cli.py`:
- Around line 523-532: Define a module-level constant for the fixed
“taxi-pulocation-borough-only” mutation ID, then replace the duplicated literal
in the taxi-inject command and _watch_taxi with that constant so both paths
remain synchronized.
In `@src/reglineage/composition.py`:
- Around line 628-655: Refactor the authorization predicate around the repair
validation into named helpers such as _observation_binds_event,
_event_is_fixed_borough, and _lease_admits_repair, separating observation,
event, delivery, and lease checks while preserving every existing condition and
fail-closed behavior. Update the validation flow to raise distinct, actionable
PermissionError messages for each failed group instead of one generic message.
- Around line 1273-1275: The repair record lookups must use the shared
REPAIR_RECIPE_VERSION instead of hardcoded v1. Update the repair_event_binding
lookup in src/reglineage/composition.py:1273-1275 and the repair_proposal lookup
in src/reglineage/mcp_backend.py:533-535, ensuring both construct IDs
consistently with propose_borough_repair and import the shared version symbol
where needed.
- Around line 862-871: The _assert_overall_active method should stop querying
the private capability_leases table through a separate sqlite3 connection. Add a
narrow public leases_by_tool(tool_id) accessor to CapabilityRuntime that
performs the runtime-owned lookup, then update _assert_overall_active to use it
for OVERALL_TOOL_ID and retain the existing ACTIVE-state validation; remove the
now-unused sqlite3 import from composition.py.
- Around line 858-860: Define a module-level _APPROVAL_VALIDITY constant set to
the one-hour timedelta, then update _unused_approval_is_current and both
approval-expiry assignments to reuse it instead of constructing separate
timedelta(hours=1) values.
- Around line 57-68: Promote the private replay.repair helpers used by
_verify_repair_candidate and _resume_activated_repair to a stable public API.
Export public equivalents for _candidate_content_hash, _governed_content_hash,
and _validation_from_json, or add one public verification function in
reglineage.replay.repair and update composition.py to call it, keeping
schema-specific candidate verification owned by that module.
In `@src/reglineage/mcp_backend.py`:
- Around line 653-662: Rename the local loaded with “repair_event_binding” from
binding to event_binding and update its proposal_hash and binding_hash checks
accordingly; rename the later “repair_publication_binding” local to
publication_binding, including its None assignment and subsequent references.
- Around line 532-539: Update the KeyError handler in the repair-proposal
loading flow to catch the exception as a named variable and raise
ToolDenied("REPAIR_NOT_PROPOSED") from that original exception, matching the
chaining pattern used by _load and _event.
In `@src/reglineage/monitoring/service.py`:
- Around line 423-463: Update bind_observation to derive the target from
mutation.previous_snapshot.target instead of PULOCATION_TARGET. Use that target
for observation lookup, expected_event_id stream_id, current snapshot retrieval,
and target-identity validation, while preserving the existing mutation binding
checks.
In `@tests/integration/test_north_star_benchmark.py`:
- Around line 172-180: Update the evidence lookup in the repair_repetitions
assertions to select the entry whose step number is 9, matching the step-number
mapping used by benchmark evidence handling, instead of indexing steps at
position 8. Keep the existing assertions unchanged after selecting the correct
step.
In `@tests/integration/test_repair_event_contract.py`:
- Around line 675-680: Add a narrowly scoped Ruff S608 suppression to the SQL
UPDATE statement in the test using PERSISTED_BOROUGH_RELATION, preserving the
constant interpolation and query behavior while leaving unrelated SQL checks
enabled.
- Around line 741-746: Replace the positional completed[3] access in the repair
test with a named binding by unpacking the repair result or using an existing
named index constant, then assert replacement_for against self.zone.lease_id
through that named reference.
- Around line 113-118: Update the changed_observation setup around
monitoring_hashed to reuse the production helper that derives the observation
identifier instead of manually combining "poll:" with canonical_sha256 and
truncation. Assert or pass the resulting identifier from that helper while
preserving the tampering test’s intended behavior.
- Around line 72-89: Update the undelivered test
test_injected_unobserved_and_observed_undelivered_fail_before_proposal to call
_inject(deliver=False) instead of duplicating the inject-and-poll flow inline.
Preserve the existing assertions and remove the now-redundant setup code; keep
_inject’s deliver branching intact.
- Around line 29-46: Update the test setup in setUp to register cleanup
callbacks immediately after creating TemporaryDirectory and TaxiSlice, so
cleanup still runs when later setup steps fail. Use addCleanup for the temporary
directory and a callback that resolves self.service at cleanup time, preserving
the existing tearDown behavior while handling the later self.service rebinding.
- Around line 652-657: Make the `patch_hash` argument type consistent in both
`publication_receipts` assertions: update the call at the shown
`publication_manifest` lookup and the corresponding call around the second
assertion to either both pass the value directly or both convert it with
`str(...)`, based on the record’s actual type.
🪄 Autofix
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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 8768cb4c-686e-47e3-9e81-11995a2773d2
📒 Files selected for processing (24)
examples/reglineage/publication-summary.jsonexamples/reglineage/runtime-attestation.jsonexamples/reglineage/verified-execution-receipt.jsonscripts/benchmark_north_star.pysrc/reglineage/cli.pysrc/reglineage/composition.pysrc/reglineage/mcp_backend.pysrc/reglineage/monitoring/service.pysrc/reglineage/monitoring/store.pysrc/reglineage/north_star_console.pysrc/reglineage/rehearsal.pysrc/reglineage/replay/models.pysrc/reglineage/replay/repair.pysrc/reglineage/runtime/store.pytests/capability/test_capability_runtime.pytests/integration/test_mcp_security_rework.pytests/integration/test_north_star_benchmark.pytests/integration/test_north_star_console.pytests/integration/test_north_star_vertical_slice.pytests/integration/test_repair_event_contract.pytests/integration/test_taxi_cli.pytests/mcp_runtime/test_protocol.pytests/monitoring/test_monitoring.pytests/replay/test_repair.py
| success = False | ||
| if ( | ||
| event is not None | ||
| and observation is not None | ||
| and service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined] | ||
| ): | ||
| service.mutator.bind_observation( # type: ignore[attr-defined] | ||
| "taxi-pulocation-borough-only", event.event_id | ||
| ) | ||
| success = service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined] | ||
| record.update( | ||
| observation_id=observation.observation_id, | ||
| observation_hash=observation.observation_hash, | ||
| disposition=observation.disposition.value, | ||
| current_snapshot_hash=observation.current_snapshot_hash, | ||
| changed_aspects=observation.changed_aspects, | ||
| event_id=event.event_id, | ||
| event_source=event.source, | ||
| event_effect=event.effect.value, | ||
| event_truth_label=event.truth_label.value, | ||
| affected_tools=event.affected_tools, | ||
| ) | ||
| if event is not None: | ||
| record["delivered"] = service.monitoring.event_delivered( # type: ignore[attr-defined] | ||
| event.event_id | ||
| ) | ||
| record["terminal_state"] = ( | ||
| "OBSERVED_DELIVERED_BOUND" | ||
| if success | ||
| else "BOUNDED_TIMEOUT" | ||
| if poll_number == max_polls | ||
| else None | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Handle bind_observation failures instead of letting the watcher crash.
service.mutator.bind_observation raises KeyError when no durable mutation record exists for "taxi-pulocation-borough-only", and raises MonitoringConflict when the observation does not bind the injected change. Both conditions are reachable here. A delivered fixed-borough event can exist without a matching mutation record, for example when an operator changes governance outside taxi-inject.
In that case _watch_taxi exits with an unhandled traceback. Every other failure path in this function emits one structured JSON record and returns a distinct exit code. Callers that parse the line-oriented record stream then receive no terminal record.
Guard the bind call and report the failure as a terminal record.
🛡️ Proposed fix to report bind failures as a terminal record
success = False
if (
event is not None
and observation is not None
and service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined]
):
- service.mutator.bind_observation( # type: ignore[attr-defined]
- "taxi-pulocation-borough-only", event.event_id
- )
- success = service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined]
+ try:
+ service.mutator.bind_observation( # type: ignore[attr-defined]
+ "taxi-pulocation-borough-only", event.event_id
+ )
+ except (KeyError, MonitoringConflict) as exc:
+ record["terminal_state"] = "OBSERVATION_NOT_BOUND"
+ record["reason"] = type(exc).__name__
+ _print_record(record)
+ return 4
+ success = service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined]
record.update(Import MonitoringConflict with the other monitoring symbols at the top of _watch_taxi:
- from reglineage.monitoring import PULOCATION_TARGET, PollDisposition
+ from reglineage.monitoring import MonitoringConflict, PULOCATION_TARGET, PollDisposition📝 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.
| success = False | |
| if ( | |
| event is not None | |
| and observation is not None | |
| and service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined] | |
| ): | |
| service.mutator.bind_observation( # type: ignore[attr-defined] | |
| "taxi-pulocation-borough-only", event.event_id | |
| ) | |
| success = service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined] | |
| record.update( | |
| observation_id=observation.observation_id, | |
| observation_hash=observation.observation_hash, | |
| disposition=observation.disposition.value, | |
| current_snapshot_hash=observation.current_snapshot_hash, | |
| changed_aspects=observation.changed_aspects, | |
| event_id=event.event_id, | |
| event_source=event.source, | |
| event_effect=event.effect.value, | |
| event_truth_label=event.truth_label.value, | |
| affected_tools=event.affected_tools, | |
| ) | |
| if event is not None: | |
| record["delivered"] = service.monitoring.event_delivered( # type: ignore[attr-defined] | |
| event.event_id | |
| ) | |
| record["terminal_state"] = ( | |
| "OBSERVED_DELIVERED_BOUND" | |
| if success | |
| else "BOUNDED_TIMEOUT" | |
| if poll_number == max_polls | |
| else None | |
| ) | |
| success = False | |
| if ( | |
| event is not None | |
| and observation is not None | |
| and service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined] | |
| ): | |
| try: | |
| service.mutator.bind_observation( # type: ignore[attr-defined] | |
| "taxi-pulocation-borough-only", event.event_id | |
| ) | |
| except (KeyError, MonitoringConflict) as exc: | |
| record["terminal_state"] = "OBSERVATION_NOT_BOUND" | |
| record["reason"] = type(exc).__name__ | |
| _print_record(record) | |
| return 4 | |
| success = service.monitoring.event_delivered(event.event_id) # type: ignore[attr-defined] | |
| record.update( | |
| observation_id=observation.observation_id, | |
| observation_hash=observation.observation_hash, | |
| disposition=observation.disposition.value, | |
| current_snapshot_hash=observation.current_snapshot_hash, | |
| changed_aspects=observation.changed_aspects, | |
| event_id=event.event_id, | |
| event_source=event.source, | |
| event_effect=event.effect.value, | |
| event_truth_label=event.truth_label.value, | |
| affected_tools=event.affected_tools, | |
| ) | |
| if event is not None: | |
| record["delivered"] = service.monitoring.event_delivered( # type: ignore[attr-defined] | |
| event.event_id | |
| ) | |
| record["terminal_state"] = ( | |
| "OBSERVED_DELIVERED_BOUND" | |
| if success | |
| else "BOUNDED_TIMEOUT" | |
| if poll_number == max_polls | |
| else None | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/reglineage/cli.py` around lines 227 - 259, Update _watch_taxi to catch
KeyError and MonitoringConflict from service.mutator.bind_observation for
"taxi-pulocation-borough-only". On failure, emit the function’s structured
terminal JSON record and return the distinct failure exit code used by its other
error paths, ensuring no subsequent unguarded processing can crash the watcher;
import MonitoringConflict alongside the existing monitoring symbols.
| existing = tuple( | ||
| item | ||
| for item in self.records.list("repair_event_binding") | ||
| if item["event_id"] == event.event_id | ||
| or item["lease_id"] == zone_lease.lease_id | ||
| ) | ||
| if existing: | ||
| if len(existing) != 1: | ||
| raise PermissionError("Repair binding is ambiguous") | ||
| binding = existing[0] | ||
| if ( | ||
| binding["event_id"] != event.event_id | ||
| or binding["event_hash"] != event.event_hash | ||
| or binding["observation_id"] != observation.observation_id | ||
| or binding["observation_hash"] != observation.observation_hash | ||
| or binding["current_snapshot_hash"] | ||
| != observation.current_snapshot_hash | ||
| or binding["lease_id"] != zone_lease.lease_id | ||
| or binding["lease_hash"] != zone_lease.lease_hash | ||
| ): | ||
| raise PermissionError("Event or lease is already bound elsewhere") |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Extract the duplicated repair-binding lookup and validation. _bound_repair_proposal and _preflight_repair_binding contain an identical records.list("repair_event_binding") filter and an identical seven-field mismatch predicate, including both error messages. repair calls both in the same invocation, so the scan runs twice and the authorization predicate exists in two copies. A future tightening applied to one copy silently leaves the other weaker.
src/reglineage/composition.py#L702-L722: replace the inline filter and mismatch check with a call to a new_existing_repair_binding(zone_lease, event, observation)helper that returns the validated binding orNone, and keep only the proposal construction and persistence here.src/reglineage/composition.py#L775-L796: replace the duplicated filter and mismatch check with the same_existing_repair_bindingcall, and keep only the proposal re-derivation and durable-record comparison here.
♻️ Proposed shared helper
def _existing_repair_binding(
self,
zone_lease: CapabilityLease,
event: ChangeEvent,
observation: PollObservation,
) -> dict[str, object] | None:
existing = tuple(
item
for item in self.records.list("repair_event_binding")
if item["event_id"] == event.event_id
or item["lease_id"] == zone_lease.lease_id
)
if not existing:
return None
if len(existing) != 1:
raise PermissionError("Repair binding is ambiguous")
binding = existing[0]
if (
binding["event_id"] != event.event_id
or binding["event_hash"] != event.event_hash
or binding["observation_id"] != observation.observation_id
or binding["observation_hash"] != observation.observation_hash
or binding["current_snapshot_hash"] != observation.current_snapshot_hash
or binding["lease_id"] != zone_lease.lease_id
or binding["lease_hash"] != zone_lease.lease_hash
):
raise PermissionError("Event or lease is already bound elsewhere")
return binding📍 Affects 1 file
src/reglineage/composition.py#L702-L722(this comment)src/reglineage/composition.py#L775-L796
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/reglineage/composition.py` around lines 702 - 722, Extract the shared
repair-binding lookup and seven-field validation into an
_existing_repair_binding(zone_lease, event, observation) helper returning the
validated binding or None, while preserving both existing PermissionError
messages. In src/reglineage/composition.py lines 702-722, replace the inline
logic with this helper and retain only proposal construction and persistence; do
the same at lines 775-796, retaining only proposal re-derivation and
durable-record comparison.
| mutation = service.mutator.inject_borough_only( | ||
| "taxi-pulocation-borough-only", | ||
| actor="external-datahub-admin", | ||
| now=_now_after(overall_lease.issued_at), | ||
| ) | ||
| steps.append({"step": 8, "status": "PROVEN", "mutation_hash": mutation.mutation_hash, "evidence_mode": "INJECTED_GOVERNANCE_EVENT"}) | ||
| polled, delivered = service.poller.poll_and_assess(service.runtime) | ||
| if polled.event is None or not service.monitoring.event_delivered(polled.event.event_id): | ||
| raise RuntimeError("Separate watcher poll did not durably deliver the injected event") | ||
| event_id = service.mutator.bind_observation( | ||
| mutation.mutation_id, polled.event.event_id | ||
| ) | ||
| observed_event = service.monitoring.event(event_id) | ||
| if observed_event.event_id != event_id or not delivered: | ||
| raise RuntimeError("Durable event lookup differs from watcher delivery") | ||
| transitions = service.runtime.transitions(zone_lease.lease_id) | ||
| steps.append({"step": 9, "status": "PROVEN", "event_id": event_id, "observation_id": polled.observation.observation_id, "observation_truth_label": polled.observation.truth_label.value, "event_truth_label": observed_event.truth_label.value, "event_source": observed_event.source, "event_effect": observed_event.effect.value, "event_delivered": service.monitoring.event_delivered(event_id), "states": tuple(item.to_state.value for item in transitions)}) | ||
|
|
||
| try: | ||
| service.repair(zone_lease, now=_now_after(overall_lease.issued_at), on_suspended=run_fallback) | ||
| service.repair(zone_lease, event_id=event_id, now=_now_after(overall_lease.issued_at), on_suspended=run_fallback) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Confirm rehearsal.py does not already clamp repair time past the observation, and compare with the clamps added in tests.
set -euo pipefail
fd -t f 'rehearsal.py' -x rg -n 'observed_at|_now_after|service\.repair' {}
echo '--- explicit clamps added elsewhere in this change set ---'
rg -n -B2 -A4 'observed_at\s*$|observed_at\b.*timedelta\(microseconds=1\)' --glob 'tests/**/*.py' --glob 'src/**/*.py'Repository: Alex-lop/RegLineage
Length of output: 1840
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- rehearsal.py relevant outline ---'
ast-grep outline src/reglineage/rehearsal.py --match 'Rehearsal' --view expanded || true
echo '--- rehearsal.py clock and repair lines ---'
sed -n '1,90p' src/reglineage/rehearsal.py
sed -n '300,390p' src/reglineage/rehearsal.py
echo '--- locate TaxiSlice.repair definitions/usages ---'
rg -n 'def repair|observed_at|Repair action must follow the exact observed event|insert_event|insert_observation|bind_observation|event_delivered' --glob 'src/**/*.py' --glob 'tests/**/*.py'Repository: Alex-lop/RegLineage
Length of output: 11429
🏁 Script executed:
Compositorically locate relevant source files and inspect only changed/relevant definitionsRepository: Alex-lop/RegLineage
Length of output: 211
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- repository files containing TaxiSlice and repair ---'
rg -n 'class TaxiSlice|def repair\(self|Repair action must follow the exact observed event|observed_at' src --glob '*.py'
echo '--- composition relevant section ---'
sed -n '1,220p' src/reglineage/composition.py
echo '--- runtime/mutator relevant sections if present ---'
sed -n '1,260p' src/reglineage/runtime.py
sed -n '1,260p' src/reglineage/mutator.py 2>/dev/null || true
sed -n '1,260p' src/reglineage/monitoring.py 2>/dev/null || trueRepository: Alex-lop/RegLineage
Length of output: 15203
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- composition.py repair implementation ---'
sed -n '800,930p' src/reglineage/composition.py
echo '--- monitoring mutator/poller event injection and observed timestamp implementation ---'
rg -n 'inject_borough_only|bind_observation|poll_and_assess|def _clock|observed_at' src/reglineage/monitoring src/reglineage/datahub src/reglineage --glob '*.py'
sed -n '760,855p' src/reglineage/monitoring/service.py 2>/dev/null || true
sed -n '1,90p' src/reglineage/monitoring/store.py 2>/dev/null || trueRepository: Alex-lop/RegLineage
Length of output: 25742
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- monitoring/service.py relevant implementation ---'
sed -n '260,430p' src/reglineage/monitoring/service.py
echo '--- datahub/replay configuration snapshot references ---'
rg -n 'clock_started_at|source_watermark|datetime.now|observed_at=|snapshot_time=' src/reglineage/replay src/reglineage/taxi_datahub.py src/reglineage/datahub src/reglineage/monitoring --glob '*.py'
sed -n '1,130p' src/reglineage/replay.py 2>/dev/null || true
sed -n '260,310p' src/reglineage/taxi_datahub.pyRepository: Alex-lop/RegLineage
Length of output: 15078
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- replay.py probe clock_started_at and source_watermark semantics ---'
sed -n '20,180p' src/reglineage/replay.py
echo '--- source_gateway clock semantics ---'
sed -n '150,210p' src/reglineage/analytics/source_gateway.py
sed -n '330,355p' src/reglineage/analytics/source_gateway.py
echo '--- deterministic behavioral probe of timestamp ordering requirements ---'
python3 - <<'PY'
from datetime import datetime, timedelta, timezone
START = datetime(2026, 8, 9, 12, tzinfo=timezone.utc)
overall_lease_issued_at = START + timedelta(seconds=80)
# Simulate observed timestamps at replay clock (not guaranteed strictly before
# the user-derived repair clock because clock_started_at <= clock if speed has advanced).
observed_at = overall_lease_issued_at + timedelta(milliseconds=1)
now = _now_after = lambda latest: (
datetime.now(timezone.utc)
if datetime.now(timezone.utc) > latest
else latest + timedelta(microseconds=1)
)
repair_now = (lambda _: max(_, _ + timedelta(microseconds=1)))(overall_lease_issued_at)
print({
"replay_clock_before_repair_now": observed_at < repair_now,
"current-rehearsal-now": repair_now.isoformat(),
"sample-observed_at": observed_at.isoformat(),
"repair-reason": "TaxiSlice.repair raises PermissionError when event.observed_at >= now or observation.observed_at >= now",
})
PYRepository: Alex-lop/RegLineage
Length of output: 297
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- locate replay and composition files ---'
git ls-files | grep -E '(^|/)replay\.py$|(^|/)rehearsal\.py$|(^|/)composition\.py$|(^|/)source_gateway\.py$|(^|/)replay/(replay|repair)\.py$' || true
echo '--- source files that exist ---'
for f in src/reglineage/replay.py src/reglineage/rehearsal.py src/reglineage/composition.py src/reglineage/replay/replay.py src/reglineage/monitoring/service.py src/reglineage/monitoring/store.py src/reglineage/analytics/source_gateway.py; do
if [ -f "$f" ]; then
echo "FOUND $f"
else
echo "MISSING $f"
fi
done
echo '--- source_gateway implementation if present ---'
if [ -f src/reglineage/analytics/source_gateway.py ]; then
sed -n '150,215p' src/reglineage/analytics/source_gateway.py
sed -n '330,395p' src/reglineage/analytics/source_gateway.py
fi
echo '--- composition repair implementation if present ---'
if [ -f src/reglineage/composition.py ]; then
sed -n '870,910p' src/reglineage/composition.py
fiRepository: Alex-lop/RegLineage
Length of output: 6596
Clamp the repair timestamp past the observation timestamps.
TaxiSlice.repair raises PermissionError("Repair action must follow the exact observed event") when event.observed_at >= now or observation.observed_at >= now. src/reglineage/rehearsal.py:373 passes _now_after(overall_lease.issued_at) as now, while the governance event/observation timestamps come from DataHub/polling and can be equal to a later clock value. Compute now as the latest of the rehearsal timestamp and the observed event/observation timestamps before calling service.repair.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/reglineage/rehearsal.py` around lines 354 - 373, Update the repair call
in the rehearsal flow after event lookup to compute a timestamp later than the
rehearsal timestamp, observed event timestamp, and polled observation timestamp,
then pass that clamped value as now to service.repair. Preserve the existing
zone_lease, event_id, and on_suspended arguments.
| event_row = conn.execute( | ||
| "SELECT payload_json FROM change_events WHERE event_id = ?", (event_id,) | ||
| "SELECT event_id, event_hash, stream_id, policy_epoch, observed_at, " | ||
| "payload_json FROM change_events WHERE event_id = ?", | ||
| (event_id,), | ||
| ).fetchone() | ||
| event = _event(event_row["payload_json"]) | ||
| event = _validated_event_row(event_row) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Handle a missing change_events row explicitly.
Line 606 passes event_row straight into _validated_event_row. If the row is absent, the helper raises TypeError: 'NoneType' object is not subscriptable instead of RuntimeContractError. The relevant-assessment check above does not guarantee the event row still exists, because a deleted change_events row leaves the impact_assessments row in place. accepted_event at line 443 and _status at line 983 both guard this case.
🛡️ Proposed fix to preserve the fail-closed contract
event_row = conn.execute(
"SELECT event_id, event_hash, stream_id, policy_epoch, observed_at, "
"payload_json FROM change_events WHERE event_id = ?",
(event_id,),
).fetchone()
+ if event_row is None:
+ raise RuntimeContractError(
+ "Revalidation requires a durable accepted event"
+ )
event = _validated_event_row(event_row)📝 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.
| event_row = conn.execute( | |
| "SELECT payload_json FROM change_events WHERE event_id = ?", (event_id,) | |
| "SELECT event_id, event_hash, stream_id, policy_epoch, observed_at, " | |
| "payload_json FROM change_events WHERE event_id = ?", | |
| (event_id,), | |
| ).fetchone() | |
| event = _event(event_row["payload_json"]) | |
| event = _validated_event_row(event_row) | |
| event_row = conn.execute( | |
| "SELECT event_id, event_hash, stream_id, policy_epoch, observed_at, " | |
| "payload_json FROM change_events WHERE event_id = ?", | |
| (event_id,), | |
| ).fetchone() | |
| if event_row is None: | |
| raise RuntimeContractError( | |
| "Revalidation requires a durable accepted event" | |
| ) | |
| event = _validated_event_row(event_row) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/reglineage/runtime/store.py` around lines 601 - 606, Update the event
lookup flow around `_validated_event_row` to explicitly check whether
`event_row` is missing and raise `RuntimeContractError` before invoking the
helper. Match the existing missing-event handling used by `accepted_event` and
`_status`, while preserving validation for present rows.
| self.now = datetime.now(timezone.utc) - timedelta(minutes=1) | ||
| self.service = TaxiSlice( | ||
| self.run_root, stage_path=self.stage_path, clock_started_at=NOW | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Do not mix wall-clock time with the fixed test clock.
self.now uses datetime.now(timezone.utc), but clock_started_at uses the fixed NOW constant at line 24. The test then advances self.now again from wall clock at lines 84-88 and 204-207. This couples assertions about chronology ordering to the machine clock and to how far NOW is from the present. When NOW moves into the past relative to real time, the relative ordering between service-clock timestamps and self.now can change, which makes the ordering tests non-deterministic over time.
Derive self.now from NOW so the whole test is deterministic.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/integration/test_repair_event_contract.py` around lines 35 - 38, Update
the test setup around self.now and the later self.now advances to derive all
timestamps from the fixed NOW constant instead of datetime.now(timezone.utc).
Preserve the existing one-minute offset and subsequent timedelta progression so
chronology assertions remain deterministic.
| with sqlite3.connect(self.service.runtime.db_path) as connection: | ||
| connection.execute( | ||
| "UPDATE change_events SET event_hash = ?, observed_at = ?, " | ||
| "payload_json = ? WHERE event_id = ?", | ||
| ( | ||
| authoritative_event.event_hash, | ||
| canonical_data(authoritative_time), | ||
| canonical_json(authoritative_event), | ||
| event_id, | ||
| ), | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Close the SQLite connection.
with sqlite3.connect(...) commits the transaction on exit, but it does not close the connection. The connection to the runtime database stays open until garbage collection. tearDown then calls self.service.close() and self.temporary.cleanup() while a handle on the database file is still open, which can fail the directory cleanup on Windows.
Wrap the connection with contextlib.closing, or close it explicitly.
🐛 Proposed fix
- with sqlite3.connect(self.service.runtime.db_path) as connection:
- connection.execute(
- "UPDATE change_events SET event_hash = ?, observed_at = ?, "
- "payload_json = ? WHERE event_id = ?",
- (
- authoritative_event.event_hash,
- canonical_data(authoritative_time),
- canonical_json(authoritative_event),
- event_id,
- ),
- )
+ with closing(sqlite3.connect(self.service.runtime.db_path)) as connection:
+ with connection:
+ connection.execute(
+ "UPDATE change_events SET event_hash = ?, observed_at = ?, "
+ "payload_json = ? WHERE event_id = ?",
+ (
+ authoritative_event.event_hash,
+ canonical_data(authoritative_time),
+ canonical_json(authoritative_event),
+ event_id,
+ ),
+ )Add the import:
-from contextlib import ExitStack
+from contextlib import ExitStack, closing📝 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.
| with sqlite3.connect(self.service.runtime.db_path) as connection: | |
| connection.execute( | |
| "UPDATE change_events SET event_hash = ?, observed_at = ?, " | |
| "payload_json = ? WHERE event_id = ?", | |
| ( | |
| authoritative_event.event_hash, | |
| canonical_data(authoritative_time), | |
| canonical_json(authoritative_event), | |
| event_id, | |
| ), | |
| ) | |
| with closing(sqlite3.connect(self.service.runtime.db_path)) as connection: | |
| with connection: | |
| connection.execute( | |
| "UPDATE change_events SET event_hash = ?, observed_at = ?, " | |
| "payload_json = ? WHERE event_id = ?", | |
| ( | |
| authoritative_event.event_hash, | |
| canonical_data(authoritative_time), | |
| canonical_json(authoritative_event), | |
| event_id, | |
| ), | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/integration/test_repair_event_contract.py` around lines 450 - 460,
Update the SQLite connection block around the change_events UPDATE to explicitly
close the connection, using contextlib.closing (and adding its import) or an
equivalent explicit close. Preserve the existing transaction commit behavior and
query parameters while ensuring the handle is released before teardown cleanup.
| def test_separate_processes_ready_inject_and_persist_delivered_history(self) -> None: | ||
| self.service.close() | ||
| self.service = None | ||
| environment = self._environment() | ||
| watcher = subprocess.Popen( | ||
| [ | ||
| sys.executable, | ||
| "-m", | ||
| "reglineage", | ||
| "taxi-watcher", | ||
| "watch", | ||
| "--interval", | ||
| "0.05", | ||
| "--max-polls", | ||
| "40", | ||
| ], | ||
| stdout=subprocess.PIPE, | ||
| stderr=subprocess.PIPE, | ||
| text=True, | ||
| env=environment, | ||
| ) | ||
| assert watcher.stdout is not None | ||
| ready_line = watcher.stdout.readline() | ||
| if not ready_line: | ||
| _, stderr = watcher.communicate(timeout=5) | ||
| self.fail(f"watcher exited before readiness: {stderr}") | ||
| ready = json.loads(ready_line) | ||
| self.assertEqual(ready["record_type"], "BASELINE_READY") | ||
|
|
||
| master, slave = pty.openpty() | ||
| injector = subprocess.Popen( | ||
| [ | ||
| sys.executable, | ||
| "-m", | ||
| "reglineage", | ||
| "taxi-inject", | ||
| "--actor", | ||
| "submitted-operator-label", | ||
| ], | ||
| stdin=slave, | ||
| stdout=slave, | ||
| stderr=slave, | ||
| env=environment, | ||
| close_fds=True, | ||
| ) | ||
| os.close(slave) | ||
| os.write(master, f"INJECT GOVERNANCE {MUTATION_ID}\n".encode()) | ||
| injector_output = bytearray() | ||
| deadline = time.monotonic() + 15 | ||
| while injector.poll() is None and time.monotonic() < deadline: | ||
| readable, _, _ = select.select((master,), (), (), 0.1) | ||
| if readable: | ||
| try: | ||
| injector_output.extend(os.read(master, 4096)) | ||
| except OSError: | ||
| break | ||
| injector.wait(timeout=5) | ||
| try: | ||
| while chunk := os.read(master, 4096): | ||
| injector_output.extend(chunk) | ||
| except OSError: | ||
| pass | ||
| finally: | ||
| os.close(master) | ||
| self.assertEqual(injector.returncode, 0, injector_output.decode(errors="replace")) | ||
| injector_text = injector_output.decode(errors="replace") | ||
| payload = json.loads( | ||
| injector_text[injector_text.index("{") : injector_text.rindex("}") + 1] | ||
| ) | ||
| self.assertEqual( | ||
| set(payload), | ||
| { | ||
| "mutation_id", | ||
| "mutation_hash", | ||
| "target_identity", | ||
| "injected_snapshot_hash", | ||
| "actor_attribution", | ||
| "truth_label", | ||
| "injected_at", | ||
| "identity_authenticated", | ||
| }, | ||
| ) | ||
| self.assertEqual(payload["actor_attribution"], "submitted-operator-label") | ||
| self.assertFalse(payload["identity_authenticated"]) | ||
| self.assertNotIn("previous_snapshot", injector_text) | ||
|
|
||
| remaining_stdout, stderr = watcher.communicate(timeout=15) | ||
| self.assertEqual(watcher.returncode, 0, stderr) | ||
| records = (ready,) + self._records(remaining_stdout) | ||
| terminal = records[-1] | ||
| self.assertEqual(terminal["terminal_state"], "OBSERVED_DELIVERED_BOUND") | ||
|
|
||
| reopened = TaxiSlice(self.run, stage_path=self.stage, clock_started_at=NOW) | ||
| try: | ||
| event_id = str(terminal["event_id"]) | ||
| self.assertTrue(reopened.monitoring.event_delivered(event_id)) | ||
| self.assertEqual(reopened.monitoring.mutation_event(MUTATION_ID), event_id) | ||
| self.assertIn( | ||
| (event_id, True), | ||
| tuple((item.event_id, delivered) for item, delivered in reopened.monitoring.events()), | ||
| ) | ||
| finally: | ||
| reopened.close() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Terminate both child processes when the test times out.
injector.wait(timeout=5) at Line 351 and watcher.communicate(timeout=15) at Line 381 raise subprocess.TimeoutExpired. Neither call kills the child on timeout. tearDown then runs self.temporary.cleanup() while the orphaned reglineage processes still hold the SQLite databases open. One timeout then produces cascading failures and leaves processes running in CI.
watcher.stdout.readline() at Line 317 also has no timeout. If the watcher hangs before it prints the readiness record, the test blocks without bound instead of failing.
Register cleanup for both processes so a timeout fails one test and releases the resources.
🛡️ Proposed fix to guarantee child-process cleanup
def test_separate_processes_ready_inject_and_persist_delivered_history(self) -> None:
self.service.close()
self.service = None
environment = self._environment()
watcher = subprocess.Popen(
[
sys.executable,
"-m",
"reglineage",
"taxi-watcher",
"watch",
"--interval",
"0.05",
"--max-polls",
"40",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=environment,
)
+ self.addCleanup(self._reap, watcher)
assert watcher.stdout is not None env=environment,
close_fds=True,
)
+ self.addCleanup(self._reap, injector)
os.close(slave)Add the helper to the test class:
`@staticmethod`
def _reap(process: subprocess.Popen) -> None:
if process.poll() is None:
process.kill()
process.wait(timeout=5)🧰 Tools
🪛 ast-grep (0.45.0)
[error] 298-314: Command coming from incoming request
Context: subprocess.Popen(
[
sys.executable,
"-m",
"reglineage",
"taxi-watcher",
"watch",
"--interval",
"0.05",
"--max-polls",
"40",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=environment,
)
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(subprocess-from-request)
[error] 324-338: Command coming from incoming request
Context: subprocess.Popen(
[
sys.executable,
"-m",
"reglineage",
"taxi-inject",
"--actor",
"submitted-operator-label",
],
stdin=slave,
stdout=slave,
stderr=slave,
env=environment,
close_fds=True,
)
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(subprocess-from-request)
🪛 Ruff (0.16.1)
[warning] 383-383: Consider (ready, *self._records(remaining_stdout)) instead of concatenation
Replace with (ready, *self._records(remaining_stdout))
(RUF005)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/integration/test_taxi_cli.py` around lines 295 - 397, Update
test_separate_processes_ready_inject_and_persist_delivered_history to guarantee
cleanup of both watcher and injector processes on every exit, using a test-class
_reap helper or equivalent registered cleanup that kills running children and
waits for them. Replace the unbounded watcher.stdout.readline readiness wait
with a bounded approach, and ensure timeout paths around injector.wait and
watcher.communicate invoke cleanup before propagating the failure.
✅ Action performedReviews resumed. |
Outcome
Resolved and merged. The
NO_GOnote on the intermediate draft applied to a chronology gap that was fixed before merge; the final head passed the required rehearsal and regression checks.What changed
Validation