Repository navigation
feat(OMN-2143): Checkpoint effect + validate compute nodes - #308
Merged
jonahgabriel merged 11 commits intoFeb 11, 2026
Merged
jonahgabriel merged 11 commits into
jonahgabriel merged 11 commits into
Conversation
…ne resilience Add two ONEX nodes for pipeline checkpoint persistence and validation: - node_checkpoint_effect (EFFECT): Write/read/list checkpoint YAML files with append-only semantics (per-attempt files, never overwrite) - node_checkpoint_validate_compute (COMPUTE): Pure structural validation of checkpoint data (path normalization, SHA format, phase consistency) - Shared models: ModelCheckpoint with discriminated union of 5 per-phase payloads - EnumCheckpointPhase: IMPLEMENT → LOCAL_REVIEW → CREATE_PR → PR_RELEASE_READY → READY_FOR_MERGE - 22 unit tests covering write/read/list round-trips and validation logic
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdd a checkpoint subsystem: phase enum, immutable checkpoint model with five discriminated phase payloads, filesystem effect node (write/read/list) with handlers and registry, compute-only validation node and handler, validation constant/exemptions updates, utility for attempt extraction, and comprehensive unit tests. Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client
participant Write as Handler_Write
participant Filesystem as Filesystem
participant Read as Handler_Read
participant Validate as Handler_Validate
Note over Client,Validate: Checkpoint lifecycle (write → read → validate)
Client->>Write: execute(checkpoint)
Write->>Filesystem: mkdir rgba(100,150,240,0.5) {base_dir}/{ticket_id}/{run_id}
Write->>Filesystem: write rgba(100,200,120,0.5) phase_<n>_<name>_a{attempt}.yaml (atomic)
Write-->>Client: checkpoint_path
Client->>Read: execute(ticket_id, run_id, phase)
Read->>Filesystem: list rgba(200,120,160,0.5) phase_*.yaml
Read->>Filesystem: read & parse latest by attempt
Read-->>Client: ModelCheckpoint
Client->>Validate: execute(checkpoint)
Validate->>Validate: run structural checks (schema, phase, shas, paths, timestamp)
Validate-->>Client: ModelCheckpointValidateOutput (is_valid, errors, warnings)
sequenceDiagram
participant Container as DI_Container
participant Registry as Registry
participant Handler as Handler
Container->>Registry: register(protocol metadata)
Container->>Handler: instantiate with container
Container->>Registry: register_handler(handler)
Registry->>Registry: validate handler methods rgba(120,180,240,0.5) (initialize, shutdown, execute)
Registry-->>Container: stored handler lookup via get_handler()
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches🧪 Generate unit tests (beta)
Comment |
![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
🤖 Fix all issues with AI agents
In
`@src/omnibase_infra/nodes/node_checkpoint_effect/handlers/handler_checkpoint_list.py`:
- Around line 111-117: The UUID conversion of run_id_raw can raise ValueError
and must be guarded: wrap the conversion in a try/except around the block that
sets run_id (the code handling run_id_raw/run_id in handler_checkpoint_list.py),
catch ValueError (and TypeError if desired), and raise a RuntimeHostError using
ModelInfraErrorContext.with_correlation(...) supplying the required
transport_type and operation parameters; include the original exception as the
cause or message so the stack/context is preserved. Ensure you reference
RuntimeHostError and ModelInfraErrorContext.with_correlation when creating the
new exception and replace the direct UUID(...) call with this guarded logic.
In
`@src/omnibase_infra/nodes/node_checkpoint_effect/handlers/handler_checkpoint_read.py`:
- Around line 154-164: Wrap the yaml.safe_load call in a try/except that catches
yaml.YAMLError, and on error raise RuntimeHostError with a
ModelInfraErrorContext created via
ModelInfraErrorContext.with_correlation(correlation_id=context.correlation_id,
transport_type=..., operation=...), including the original YAMLError message in
the RuntimeHostError message and chaining from the caught exception; similarly
ensure the existing ValidationError branch also constructs its context via
ModelInfraErrorContext.with_correlation and passes the transport_type and
operation parameters when raising RuntimeHostError from
ModelCheckpoint.model_validate.
- Around line 136-153: The current lexicographic sort of matching_files can
misorder attempts like "a10" vs "a2"; update the sorting so it extracts the
numeric attempt portion from filenames and sorts numerically: in the handler
(handler_checkpoint_read) where matching_files = sorted(..., key=lambda p:
p.name), replace the key with a function that uses a regex to capture the
attempt number after phase_prefix (or parses the substring), convert it to int
for ordering, and fall back to p.name or 0 if no number is found so the true
latest attempt is selected deterministically.
- Around line 103-114: Wrap the parsing of phase_raw and run_id so that any
ValueError is caught and re-raised as a RuntimeHostError using a
ModelInfraErrorContext created by ModelInfraErrorContext.with_correlation(...,
transport_type=..., operation=...), i.e. around EnumCheckpointPhase(phase_raw)
and UUID(str(run_id)) ensure you catch ValueError and raise
RuntimeHostError(error_message,
context=ModelInfraErrorContext.with_correlation(correlation_id,
transport_type=<appropriate>, operation=<appropriate>)) while keeping the
existing branch logic for when phase_raw is already an EnumCheckpointPhase or
run_id is already a UUID; reference the symbols EnumCheckpointPhase, UUID,
RuntimeHostError, and ModelInfraErrorContext.with_correlation when making the
change.
In
`@src/omnibase_infra/nodes/node_checkpoint_validate_compute/handlers/handler_checkpoint_validate.py`:
- Around line 76-81: The current code sets corr_id to the zero UUID when
envelope["correlation_id"] is missing or not a UUID; instead, update the logic
in handler_checkpoint_validate (around variables correlation_id_raw, corr_id) to
parse correlation_id if it is a UUID instance or a valid UUID string, and
otherwise generate a new UUID using uuid4(); ensure you import uuid4 from uuid
and use the generated uuid for corr_id and include it in all subsequent
error/context logging to preserve traceability.
In
`@tests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_read.py`:
- Around line 46-203: Add the pytest unit marker to each async test in the
TestHandlerCheckpointRead class: decorate test_read_returns_written_checkpoint,
test_read_returns_latest_attempt, test_read_missing_directory_returns_error, and
test_read_missing_phase_returns_error with `@pytest.mark.unit` (import pytest if
not present) so they are classified correctly; keep markers consistent with
coding guidelines (use `@pytest.mark.unit` for these unit tests) and ensure the
decorator is applied directly above each test coroutine definition.
In
`@tests/unit/nodes/test_node_checkpoint_validate_compute/test_handler_checkpoint_validate.py`:
- Around line 116-138: The test named test_phase_payload_mismatch_is_error is
misleading because it constructs a checkpoint with header
EnumCheckpointPhase.LOCAL_REVIEW and a matching ModelPhasePayloadLocalReview and
asserts validity; either rename the test to something like
test_phase_payload_match_passes (update the test function name and any
docstring) or add a new test (e.g., test_phase_payload_mismatch_is_error) that
uses _valid_checkpoint with a mismatched payload (e.g., header
EnumCheckpointPhase.LOCAL_REVIEW but phase_payload set to a different payload
type such as ModelPhasePayloadImplement) and assert result.result.is_valid is
False after calling HandlerCheckpointValidate.initialize and .execute(env);
locate and update the test function name test_phase_payload_mismatch_is_error
and references to _valid_checkpoint, HandlerCheckpointValidate,
EnumCheckpointPhase.LOCAL_REVIEW, and
ModelPhasePayloadLocalReview/ModelPhasePayloadImplement accordingly.
🧹 Nitpick comments (6)
tests/unit/nodes/test_node_checkpoint_validate_compute/test_handler_checkpoint_validate.py (1)
58-59: Add@pytest.mark.unitmarker to the test class.The test class is missing the required pytest marker for test classification. As per coding guidelines, tests should use pytest markers for classification.
Proposed fix
+@pytest.mark.unit class TestHandlerCheckpointValidate: """Tests for HandlerCheckpointValidate."""src/omnibase_infra/nodes/node_checkpoint_effect/contract.yaml (1)
22-34: Consider documentingcorrelation_idanderrorin output_fields.The
io_operationsoutput_fields don't includecorrelation_idanderror, which are present inModelCheckpointEffectOutput(as shown in the relevant code snippets). While this may be intentional to show only operation-specific fields, consider adding these common response fields for completeness, or document that they're implicitly included.Proposed addition
io_operations: - name: "write_checkpoint" description: "Write a phase checkpoint YAML to disk (append-only)." input_fields: ["checkpoint"] - output_fields: ["success", "checkpoint_path"] + output_fields: ["success", "correlation_id", "checkpoint_path", "error"] - name: "read_checkpoint" description: "Read a specific checkpoint from disk." input_fields: ["ticket_id", "run_id", "phase"] - output_fields: ["success", "checkpoint"] + output_fields: ["success", "correlation_id", "checkpoint", "error"] - name: "list_checkpoints" description: "List available checkpoints for a ticket/run." input_fields: ["ticket_id", "run_id"] - output_fields: ["success", "checkpoints"] + output_fields: ["success", "correlation_id", "checkpoints", "error"]src/omnibase_infra/models/checkpoint/model_phase_payload_create_pr.py (1)
26-30: Consider adding URL format validation forpr_url.The
pr_urlfield only enforcesmin_length=1. Consider adding a regex pattern or using Pydantic'sHttpUrltype to validate that the value is a well-formed URL, preventing invalid data from being persisted.Option 1: Use Pydantic HttpUrl
-from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, HttpUrl class ModelPhasePayloadCreatePr(BaseModel): """Payload captured after the create_pr phase completes.""" model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) phase: Literal["create_pr"] = Field( default="create_pr", description="Discriminator field for phase payload union.", ) - pr_url: str = Field( + pr_url: HttpUrl = Field( ..., - min_length=1, description="Full URL of the created pull request.", )Option 2: Add pattern validation
pr_url: str = Field( ..., min_length=1, + pattern=r"^https?://", description="Full URL of the created pull request.", )tests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_write.py (1)
56-57: Add@pytest.mark.unitmarker to the test class.Per coding guidelines, test files should use pytest markers for classification. This test class is missing the
@pytest.mark.unitmarker.♻️ Proposed fix
+@pytest.mark.unit class TestHandlerCheckpointWrite: """Tests for HandlerCheckpointWrite."""As per coding guidelines: "Use pytest markers:
@pytest.mark.unit,@pytest.mark.integration,@pytest.mark.slow,@pytest.mark.chaos,@pytest.mark.serialfor test classification"src/omnibase_infra/nodes/node_checkpoint_effect/handlers/handler_checkpoint_write.py (1)
97-107: Minor redundancy in correlation_id handling.Line 107
corr_id = context.correlation_id or uuid4()is redundant sinceModelInfraErrorContext.with_correlation()already auto-generates a UUID whenNoneis passed (see relevant snippet at lines 66-95 ofmodel_infra_error_context.py). Thecontext.correlation_idwill never beNoneafter that call.♻️ Suggested simplification
# with_correlation() auto-generates a UUID when None is passed - corr_id = context.correlation_id or uuid4() + corr_id = context.correlation_id # Guaranteed non-None by with_correlation()You may want to add a type assertion if mypy complains:
assert context.correlation_id is not None # Guaranteed by with_correlation() corr_id = context.correlation_idtests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_list.py (1)
51-52: Add@pytest.mark.unitmarker to the test class.Per coding guidelines, test files should use pytest markers for classification.
♻️ Proposed fix
+@pytest.mark.unit class TestHandlerCheckpointList: """Tests for HandlerCheckpointList."""As per coding guidelines: "Use pytest markers:
@pytest.mark.unit,@pytest.mark.integration,@pytest.mark.slow,@pytest.mark.chaos,@pytest.mark.serialfor test classification"
Comment on lines
+111
to
+117
| run_id_raw = envelope.get("run_id") | ||
| if run_id_raw is not None: | ||
| run_id = ( | ||
| UUID(str(run_id_raw)) | ||
| if not isinstance(run_id_raw, UUID) | ||
| else run_id_raw | ||
| ) |
There was a problem hiding this comment.
Guard invalid run_id parsing with structured RuntimeHostError.
UUID conversion can raise ValueError and should be wrapped with the existing context.
🐛 Proposed fix
run_id_raw = envelope.get("run_id")
if run_id_raw is not None:
- run_id = (
- UUID(str(run_id_raw))
- if not isinstance(run_id_raw, UUID)
- else run_id_raw
- )
+ try:
+ run_id = (
+ UUID(str(run_id_raw))
+ if not isinstance(run_id_raw, UUID)
+ else run_id_raw
+ )
+ except (ValueError, TypeError) as exc:
+ raise RuntimeHostError(
+ "list_checkpoints requires a valid run_id UUID",
+ context=context,
+ ) from exc🤖 Prompt for AI Agents
In
`@src/omnibase_infra/nodes/node_checkpoint_effect/handlers/handler_checkpoint_list.py`
around lines 111 - 117, The UUID conversion of run_id_raw can raise ValueError
and must be guarded: wrap the conversion in a try/except around the block that
sets run_id (the code handling run_id_raw/run_id in handler_checkpoint_list.py),
catch ValueError (and TypeError if desired), and raise a RuntimeHostError using
ModelInfraErrorContext.with_correlation(...) supplying the required
transport_type and operation parameters; include the original exception as the
cause or message so the stack/context is preserved. Ensure you reference
RuntimeHostError and ModelInfraErrorContext.with_correlation when creating the
new exception and replace the direct UUID(...) call with this guarded logic.
Comment on lines
+103
to
+114
| # Coerce phase | ||
| if isinstance(phase_raw, EnumCheckpointPhase): | ||
| phase = phase_raw | ||
| elif isinstance(phase_raw, str): | ||
| phase = EnumCheckpointPhase(phase_raw) | ||
| else: | ||
| raise RuntimeHostError( | ||
| f"Unexpected phase type: {type(phase_raw).__name__}", | ||
| context=context, | ||
| ) | ||
|
|
||
| run_id_val = UUID(str(run_id)) if not isinstance(run_id, UUID) else run_id |
There was a problem hiding this comment.
Wrap invalid phase/run_id parsing with structured RuntimeHostError.
EnumCheckpointPhase(...) and UUID(...) can raise ValueError, which currently bubbles without context.
🐛 Proposed fix
if isinstance(phase_raw, EnumCheckpointPhase):
phase = phase_raw
elif isinstance(phase_raw, str):
- phase = EnumCheckpointPhase(phase_raw)
+ try:
+ phase = EnumCheckpointPhase(phase_raw)
+ except ValueError as exc:
+ raise RuntimeHostError(
+ f"Unknown phase value: {phase_raw}",
+ context=context,
+ ) from exc
else:
raise RuntimeHostError(
f"Unexpected phase type: {type(phase_raw).__name__}",
context=context,
)
- run_id_val = UUID(str(run_id)) if not isinstance(run_id, UUID) else run_id
+ try:
+ run_id_val = UUID(str(run_id)) if not isinstance(run_id, UUID) else run_id
+ except (ValueError, TypeError) as exc:
+ raise RuntimeHostError(
+ "read_checkpoint requires a valid run_id UUID",
+ context=context,
+ ) from exc📝 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.
Suggested change
| # Coerce phase | |
| if isinstance(phase_raw, EnumCheckpointPhase): | |
| phase = phase_raw | |
| elif isinstance(phase_raw, str): | |
| phase = EnumCheckpointPhase(phase_raw) | |
| else: | |
| raise RuntimeHostError( | |
| f"Unexpected phase type: {type(phase_raw).__name__}", | |
| context=context, | |
| ) | |
| run_id_val = UUID(str(run_id)) if not isinstance(run_id, UUID) else run_id | |
| # Coerce phase | |
| if isinstance(phase_raw, EnumCheckpointPhase): | |
| phase = phase_raw | |
| elif isinstance(phase_raw, str): | |
| try: | |
| phase = EnumCheckpointPhase(phase_raw) | |
| except ValueError as exc: | |
| raise RuntimeHostError( | |
| f"Unknown phase value: {phase_raw}", | |
| context=ModelInfraErrorContext.with_correlation( | |
| correlation_id=..., # propagate from incoming request | |
| transport_type=..., | |
| operation="checkpoint_read" | |
| ), | |
| ) from exc | |
| else: | |
| raise RuntimeHostError( | |
| f"Unexpected phase type: {type(phase_raw).__name__}", | |
| context=context, | |
| ) | |
| try: | |
| run_id_val = UUID(str(run_id)) if not isinstance(run_id, UUID) else run_id | |
| except (ValueError, TypeError) as exc: | |
| raise RuntimeHostError( | |
| "read_checkpoint requires a valid run_id UUID", | |
| context=ModelInfraErrorContext.with_correlation( | |
| correlation_id=..., | |
| transport_type=..., | |
| operation="checkpoint_read" | |
| ), | |
| ) from exc |
🤖 Prompt for AI Agents
In
`@src/omnibase_infra/nodes/node_checkpoint_effect/handlers/handler_checkpoint_read.py`
around lines 103 - 114, Wrap the parsing of phase_raw and run_id so that any
ValueError is caught and re-raised as a RuntimeHostError using a
ModelInfraErrorContext created by ModelInfraErrorContext.with_correlation(...,
transport_type=..., operation=...), i.e. around EnumCheckpointPhase(phase_raw)
and UUID(str(run_id)) ensure you catch ValueError and raise
RuntimeHostError(error_message,
context=ModelInfraErrorContext.with_correlation(correlation_id,
transport_type=<appropriate>, operation=<appropriate>)) while keeping the
existing branch logic for when phase_raw is already an EnumCheckpointPhase or
run_id is already a UUID; reference the symbols EnumCheckpointPhase, UUID,
RuntimeHostError, and ModelInfraErrorContext.with_correlation when making the
change.
Comment on lines
+154
to
+164
| # Read the latest attempt (last file alphabetically = highest attempt) | ||
| latest_file = matching_files[-1] | ||
| raw_data = yaml.safe_load(latest_file.read_text(encoding="utf-8")) | ||
|
|
||
| try: | ||
| checkpoint = ModelCheckpoint.model_validate(raw_data) | ||
| except ValidationError as exc: | ||
| raise RuntimeHostError( | ||
| f"Corrupt checkpoint file {latest_file.name}: {exc.error_count()} errors", | ||
| context=context, | ||
| ) from exc |
There was a problem hiding this comment.
Handle YAML parse failures with structured errors.
yaml.safe_load can raise YAMLError, which currently escapes without context.
🐛 Proposed fix
- raw_data = yaml.safe_load(latest_file.read_text(encoding="utf-8"))
+ try:
+ raw_data = yaml.safe_load(latest_file.read_text(encoding="utf-8"))
+ except yaml.YAMLError as exc:
+ raise RuntimeHostError(
+ f"Failed to parse checkpoint YAML {latest_file.name}",
+ context=context,
+ ) from exc🤖 Prompt for AI Agents
In
`@src/omnibase_infra/nodes/node_checkpoint_effect/handlers/handler_checkpoint_read.py`
around lines 154 - 164, Wrap the yaml.safe_load call in a try/except that
catches yaml.YAMLError, and on error raise RuntimeHostError with a
ModelInfraErrorContext created via
ModelInfraErrorContext.with_correlation(correlation_id=context.correlation_id,
transport_type=..., operation=...), including the original YAMLError message in
the RuntimeHostError message and chaining from the caught exception; similarly
ensure the existing ValidationError branch also constructs its context via
ModelInfraErrorContext.with_correlation and passes the transport_type and
operation parameters when raising RuntimeHostError from
ModelCheckpoint.model_validate.
Comment on lines
+76
to
+81
| correlation_id_raw = envelope.get("correlation_id") | ||
| corr_id = ( | ||
| correlation_id_raw | ||
| if isinstance(correlation_id_raw, UUID) | ||
| else UUID("00000000-0000-0000-0000-000000000000") | ||
| ) |
There was a problem hiding this comment.
Generate/parse correlation_id instead of using the zero UUID.
A missing or non-UUID input should result in a real uuid4() to preserve traceability.
🐛 Proposed fix
correlation_id_raw = envelope.get("correlation_id")
- corr_id = (
- correlation_id_raw
- if isinstance(correlation_id_raw, UUID)
- else UUID("00000000-0000-0000-0000-000000000000")
- )
+ try:
+ if isinstance(correlation_id_raw, UUID):
+ corr_id = correlation_id_raw
+ elif correlation_id_raw:
+ corr_id = UUID(str(correlation_id_raw))
+ else:
+ corr_id = uuid4()
+ except (ValueError, TypeError):
+ corr_id = uuid4()🤖 Prompt for AI Agents
In
`@src/omnibase_infra/nodes/node_checkpoint_validate_compute/handlers/handler_checkpoint_validate.py`
around lines 76 - 81, The current code sets corr_id to the zero UUID when
envelope["correlation_id"] is missing or not a UUID; instead, update the logic
in handler_checkpoint_validate (around variables correlation_id_raw, corr_id) to
parse correlation_id if it is a UUID instance or a valid UUID string, and
otherwise generate a new UUID using uuid4(); ensure you import uuid4 from uuid
and use the generated uuid for corr_id and include it in all subsequent
error/context logging to preserve traceability.
Comment on lines
+46
to
+203
| class TestHandlerCheckpointRead: | ||
| """Tests for HandlerCheckpointRead.""" | ||
|
|
||
| async def test_read_returns_written_checkpoint( | ||
| self, | ||
| writer: HandlerCheckpointWrite, | ||
| reader: HandlerCheckpointRead, | ||
| tmp_path: Path, | ||
| ) -> None: | ||
| """Round-trip: write then read returns the same checkpoint.""" | ||
| await writer.initialize({}) | ||
| await reader.initialize({}) | ||
|
|
||
| run_id = uuid4() | ||
| checkpoint = ModelCheckpoint( | ||
| run_id=run_id, | ||
| ticket_id="OMN-2143", | ||
| phase=EnumCheckpointPhase.IMPLEMENT, | ||
| timestamp_utc=datetime.now(UTC), | ||
| repo_commit_map={"infra": "deadbeef"}, | ||
| attempt_number=1, | ||
| phase_payload=ModelPhasePayloadImplement( | ||
| branch_name="feature-branch", | ||
| commit_sha="deadbeef", | ||
| files_changed=("src/foo.py",), | ||
| ), | ||
| ) | ||
|
|
||
| # Write | ||
| write_env: dict[str, object] = { | ||
| "checkpoint": checkpoint, | ||
| "correlation_id": uuid4(), | ||
| "base_dir": str(tmp_path), | ||
| } | ||
| await writer.execute(write_env) | ||
|
|
||
| # Read | ||
| read_env: dict[str, object] = { | ||
| "ticket_id": "OMN-2143", | ||
| "run_id": run_id, | ||
| "phase": EnumCheckpointPhase.IMPLEMENT, | ||
| "correlation_id": uuid4(), | ||
| "base_dir": str(tmp_path), | ||
| } | ||
| result = await reader.execute(read_env) | ||
| output = result.result | ||
|
|
||
| assert output.success is True | ||
| assert output.checkpoint is not None | ||
| assert output.checkpoint.ticket_id == "OMN-2143" | ||
| assert output.checkpoint.repo_commit_map == {"infra": "deadbeef"} | ||
|
|
||
| async def test_read_returns_latest_attempt( | ||
| self, | ||
| writer: HandlerCheckpointWrite, | ||
| reader: HandlerCheckpointRead, | ||
| tmp_path: Path, | ||
| ) -> None: | ||
| """When multiple attempts exist, read returns the highest attempt.""" | ||
| await writer.initialize({}) | ||
| await reader.initialize({}) | ||
|
|
||
| run_id = uuid4() | ||
|
|
||
| for attempt in (1, 2, 3): | ||
| cp = ModelCheckpoint( | ||
| run_id=run_id, | ||
| ticket_id="OMN-2143", | ||
| phase=EnumCheckpointPhase.LOCAL_REVIEW, | ||
| timestamp_utc=datetime.now(UTC), | ||
| attempt_number=attempt, | ||
| phase_payload={ | ||
| "phase": "local_review", | ||
| "iteration_count": attempt, | ||
| "issue_fingerprints": [], | ||
| "last_clean_sha": f"sha{attempt:04d}aa", | ||
| }, | ||
| ) | ||
| write_env: dict[str, object] = { | ||
| "checkpoint": cp, | ||
| "correlation_id": uuid4(), | ||
| "base_dir": str(tmp_path), | ||
| } | ||
| await writer.execute(write_env) | ||
|
|
||
| # Read should return attempt 3 | ||
| read_env: dict[str, object] = { | ||
| "ticket_id": "OMN-2143", | ||
| "run_id": run_id, | ||
| "phase": EnumCheckpointPhase.LOCAL_REVIEW, | ||
| "correlation_id": uuid4(), | ||
| "base_dir": str(tmp_path), | ||
| } | ||
| result = await reader.execute(read_env) | ||
| assert result.result.success is True | ||
| assert result.result.checkpoint.attempt_number == 3 | ||
|
|
||
| async def test_read_missing_directory_returns_error( | ||
| self, | ||
| reader: HandlerCheckpointRead, | ||
| tmp_path: Path, | ||
| ) -> None: | ||
| """Read returns failure when the checkpoint directory does not exist.""" | ||
| await reader.initialize({}) | ||
|
|
||
| read_env: dict[str, object] = { | ||
| "ticket_id": "OMN-NOEXIST", | ||
| "run_id": uuid4(), | ||
| "phase": EnumCheckpointPhase.IMPLEMENT, | ||
| "correlation_id": uuid4(), | ||
| "base_dir": str(tmp_path), | ||
| } | ||
| result = await reader.execute(read_env) | ||
| assert result.result.success is False | ||
| assert "not found" in result.result.error | ||
|
|
||
| async def test_read_missing_phase_returns_error( | ||
| self, | ||
| writer: HandlerCheckpointWrite, | ||
| reader: HandlerCheckpointRead, | ||
| tmp_path: Path, | ||
| ) -> None: | ||
| """Read returns failure when the requested phase has no checkpoints.""" | ||
| await writer.initialize({}) | ||
| await reader.initialize({}) | ||
|
|
||
| run_id = uuid4() | ||
| # Write implement phase | ||
| cp = ModelCheckpoint( | ||
| run_id=run_id, | ||
| ticket_id="OMN-2143", | ||
| phase=EnumCheckpointPhase.IMPLEMENT, | ||
| timestamp_utc=datetime.now(UTC), | ||
| attempt_number=1, | ||
| phase_payload=ModelPhasePayloadImplement( | ||
| branch_name="branch", | ||
| commit_sha="abc1234", | ||
| ), | ||
| ) | ||
| await writer.execute( | ||
| { | ||
| "checkpoint": cp, | ||
| "correlation_id": uuid4(), | ||
| "base_dir": str(tmp_path), | ||
| } | ||
| ) | ||
|
|
||
| # Try to read a different phase | ||
| read_env: dict[str, object] = { | ||
| "ticket_id": "OMN-2143", | ||
| "run_id": run_id, | ||
| "phase": EnumCheckpointPhase.CREATE_PR, | ||
| "correlation_id": uuid4(), | ||
| "base_dir": str(tmp_path), | ||
| } | ||
| result = await reader.execute(read_env) | ||
| assert result.result.success is False | ||
| assert "No checkpoint found" in result.result.error |
There was a problem hiding this comment.
Add unit markers so these tests are classified.
These async tests are missing pytest markers, which breaks the required test classification.
✅ Proposed fix
+@pytest.mark.unit
class TestHandlerCheckpointRead:📝 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.
Suggested change
| class TestHandlerCheckpointRead: | |
| """Tests for HandlerCheckpointRead.""" | |
| async def test_read_returns_written_checkpoint( | |
| self, | |
| writer: HandlerCheckpointWrite, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """Round-trip: write then read returns the same checkpoint.""" | |
| await writer.initialize({}) | |
| await reader.initialize({}) | |
| run_id = uuid4() | |
| checkpoint = ModelCheckpoint( | |
| run_id=run_id, | |
| ticket_id="OMN-2143", | |
| phase=EnumCheckpointPhase.IMPLEMENT, | |
| timestamp_utc=datetime.now(UTC), | |
| repo_commit_map={"infra": "deadbeef"}, | |
| attempt_number=1, | |
| phase_payload=ModelPhasePayloadImplement( | |
| branch_name="feature-branch", | |
| commit_sha="deadbeef", | |
| files_changed=("src/foo.py",), | |
| ), | |
| ) | |
| # Write | |
| write_env: dict[str, object] = { | |
| "checkpoint": checkpoint, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| await writer.execute(write_env) | |
| # Read | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-2143", | |
| "run_id": run_id, | |
| "phase": EnumCheckpointPhase.IMPLEMENT, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| output = result.result | |
| assert output.success is True | |
| assert output.checkpoint is not None | |
| assert output.checkpoint.ticket_id == "OMN-2143" | |
| assert output.checkpoint.repo_commit_map == {"infra": "deadbeef"} | |
| async def test_read_returns_latest_attempt( | |
| self, | |
| writer: HandlerCheckpointWrite, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """When multiple attempts exist, read returns the highest attempt.""" | |
| await writer.initialize({}) | |
| await reader.initialize({}) | |
| run_id = uuid4() | |
| for attempt in (1, 2, 3): | |
| cp = ModelCheckpoint( | |
| run_id=run_id, | |
| ticket_id="OMN-2143", | |
| phase=EnumCheckpointPhase.LOCAL_REVIEW, | |
| timestamp_utc=datetime.now(UTC), | |
| attempt_number=attempt, | |
| phase_payload={ | |
| "phase": "local_review", | |
| "iteration_count": attempt, | |
| "issue_fingerprints": [], | |
| "last_clean_sha": f"sha{attempt:04d}aa", | |
| }, | |
| ) | |
| write_env: dict[str, object] = { | |
| "checkpoint": cp, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| await writer.execute(write_env) | |
| # Read should return attempt 3 | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-2143", | |
| "run_id": run_id, | |
| "phase": EnumCheckpointPhase.LOCAL_REVIEW, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| assert result.result.success is True | |
| assert result.result.checkpoint.attempt_number == 3 | |
| async def test_read_missing_directory_returns_error( | |
| self, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """Read returns failure when the checkpoint directory does not exist.""" | |
| await reader.initialize({}) | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-NOEXIST", | |
| "run_id": uuid4(), | |
| "phase": EnumCheckpointPhase.IMPLEMENT, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| assert result.result.success is False | |
| assert "not found" in result.result.error | |
| async def test_read_missing_phase_returns_error( | |
| self, | |
| writer: HandlerCheckpointWrite, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """Read returns failure when the requested phase has no checkpoints.""" | |
| await writer.initialize({}) | |
| await reader.initialize({}) | |
| run_id = uuid4() | |
| # Write implement phase | |
| cp = ModelCheckpoint( | |
| run_id=run_id, | |
| ticket_id="OMN-2143", | |
| phase=EnumCheckpointPhase.IMPLEMENT, | |
| timestamp_utc=datetime.now(UTC), | |
| attempt_number=1, | |
| phase_payload=ModelPhasePayloadImplement( | |
| branch_name="branch", | |
| commit_sha="abc1234", | |
| ), | |
| ) | |
| await writer.execute( | |
| { | |
| "checkpoint": cp, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| ) | |
| # Try to read a different phase | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-2143", | |
| "run_id": run_id, | |
| "phase": EnumCheckpointPhase.CREATE_PR, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| assert result.result.success is False | |
| assert "No checkpoint found" in result.result.error | |
| `@pytest.mark.unit` | |
| class TestHandlerCheckpointRead: | |
| """Tests for HandlerCheckpointRead.""" | |
| async def test_read_returns_written_checkpoint( | |
| self, | |
| writer: HandlerCheckpointWrite, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """Round-trip: write then read returns the same checkpoint.""" | |
| await writer.initialize({}) | |
| await reader.initialize({}) | |
| run_id = uuid4() | |
| checkpoint = ModelCheckpoint( | |
| run_id=run_id, | |
| ticket_id="OMN-2143", | |
| phase=EnumCheckpointPhase.IMPLEMENT, | |
| timestamp_utc=datetime.now(UTC), | |
| repo_commit_map={"infra": "deadbeef"}, | |
| attempt_number=1, | |
| phase_payload=ModelPhasePayloadImplement( | |
| branch_name="feature-branch", | |
| commit_sha="deadbeef", | |
| files_changed=("src/foo.py",), | |
| ), | |
| ) | |
| # Write | |
| write_env: dict[str, object] = { | |
| "checkpoint": checkpoint, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| await writer.execute(write_env) | |
| # Read | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-2143", | |
| "run_id": run_id, | |
| "phase": EnumCheckpointPhase.IMPLEMENT, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| output = result.result | |
| assert output.success is True | |
| assert output.checkpoint is not None | |
| assert output.checkpoint.ticket_id == "OMN-2143" | |
| assert output.checkpoint.repo_commit_map == {"infra": "deadbeef"} | |
| async def test_read_returns_latest_attempt( | |
| self, | |
| writer: HandlerCheckpointWrite, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """When multiple attempts exist, read returns the highest attempt.""" | |
| await writer.initialize({}) | |
| await reader.initialize({}) | |
| run_id = uuid4() | |
| for attempt in (1, 2, 3): | |
| cp = ModelCheckpoint( | |
| run_id=run_id, | |
| ticket_id="OMN-2143", | |
| phase=EnumCheckpointPhase.LOCAL_REVIEW, | |
| timestamp_utc=datetime.now(UTC), | |
| attempt_number=attempt, | |
| phase_payload={ | |
| "phase": "local_review", | |
| "iteration_count": attempt, | |
| "issue_fingerprints": [], | |
| "last_clean_sha": f"sha{attempt:04d}aa", | |
| }, | |
| ) | |
| write_env: dict[str, object] = { | |
| "checkpoint": cp, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| await writer.execute(write_env) | |
| # Read should return attempt 3 | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-2143", | |
| "run_id": run_id, | |
| "phase": EnumCheckpointPhase.LOCAL_REVIEW, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| assert result.result.success is True | |
| assert result.result.checkpoint.attempt_number == 3 | |
| async def test_read_missing_directory_returns_error( | |
| self, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """Read returns failure when the checkpoint directory does not exist.""" | |
| await reader.initialize({}) | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-NOEXIST", | |
| "run_id": uuid4(), | |
| "phase": EnumCheckpointPhase.IMPLEMENT, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| assert result.result.success is False | |
| assert "not found" in result.result.error | |
| async def test_read_missing_phase_returns_error( | |
| self, | |
| writer: HandlerCheckpointWrite, | |
| reader: HandlerCheckpointRead, | |
| tmp_path: Path, | |
| ) -> None: | |
| """Read returns failure when the requested phase has no checkpoints.""" | |
| await writer.initialize({}) | |
| await reader.initialize({}) | |
| run_id = uuid4() | |
| # Write implement phase | |
| cp = ModelCheckpoint( | |
| run_id=run_id, | |
| ticket_id="OMN-2143", | |
| phase=EnumCheckpointPhase.IMPLEMENT, | |
| timestamp_utc=datetime.now(UTC), | |
| attempt_number=1, | |
| phase_payload=ModelPhasePayloadImplement( | |
| branch_name="branch", | |
| commit_sha="abc1234", | |
| ), | |
| ) | |
| await writer.execute( | |
| { | |
| "checkpoint": cp, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| ) | |
| # Try to read a different phase | |
| read_env: dict[str, object] = { | |
| "ticket_id": "OMN-2143", | |
| "run_id": run_id, | |
| "phase": EnumCheckpointPhase.CREATE_PR, | |
| "correlation_id": uuid4(), | |
| "base_dir": str(tmp_path), | |
| } | |
| result = await reader.execute(read_env) | |
| assert result.result.success is False | |
| assert "No checkpoint found" in result.result.error |
🤖 Prompt for AI Agents
In `@tests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_read.py`
around lines 46 - 203, Add the pytest unit marker to each async test in the
TestHandlerCheckpointRead class: decorate test_read_returns_written_checkpoint,
test_read_returns_latest_attempt, test_read_missing_directory_returns_error, and
test_read_missing_phase_returns_error with `@pytest.mark.unit` (import pytest if
not present) so they are classified correctly; keep markers consistent with
coding guidelines (use `@pytest.mark.unit` for these unit tests) and ensure the
decorator is applied directly above each test coroutine definition.
… handling, dead code - Write handler now refuses to overwrite existing checkpoints (append-only guard) - Read handler catches yaml.YAMLError and rejects non-dict YAML content - List handler skips empty/non-mapping YAML files instead of crashing - Removed redundant `or uuid4()` fallback (with_correlation guarantees UUID) - Fixed __all__ alphabetical ordering in enums/__init__.py - Added handler classification docstring to HandlerCheckpointValidate - Added 4 new tests covering overwrite guard, invalid YAML, and empty files Review iteration: 1/10
…t sort - All 3 effect handlers now validate resolved paths stay under base_dir via is_relative_to() to prevent ticket_id path traversal attacks - Read handler sorts checkpoint files by extracted attempt number instead of lexicographic filename order (fixes incorrect result for attempt >= 10) - Added 3 new tests: path traversal write/read, numeric sort verification Review iteration: 2/10
…on hardening fixes - Atomic checkpoint write via tempfile + os.link(O_CREAT-equivalent) prevents partial writes on crash and eliminates TOCTOU race on duplicate detection - timestamp_utc field_validator rejects naive datetimes, normalizes to UTC - List handler sorts by numeric attempt_number (consistent with read handler) - Operation field uses Literal type for compile-time validation - Removed dead naive-datetime branch in validate handler Review iteration: 1/10
… hex pattern + UTC validator + dedup - Chain FileExistsError via `from exc` to preserve traceback - Assert corr_id is not None in all 3 effect handlers (mypy type narrowing) - Add hex pattern constraint (^[0-9a-f]+$) to SHA fields across all payload models - Add _ensure_utc validator to ModelPhasePayloadReadyForMerge.label_applied_at - Extract _attempt_number to module level in read/list handlers (eliminate per-call overhead) - Fix test fixture using non-hex SHA values Review iteration: 2/10
…zation, dedup helper - Replace fragile list().index() with explicit _PHASE_ORDINALS mapping for stable checkpoint filenames across enum reordering - Add symlink escape guard on scan_dirs in list handler - Log relative_path instead of full absolute path in write handler - Deduplicate _attempt_number: single definition in list handler, imported by read - Fix stale "alphabetically" comment in read handler Review iteration: 3/10
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In
`@src/omnibase_infra/nodes/node_checkpoint_validate_compute/handlers/handler_checkpoint_validate.py`:
- Around line 163-166: The current loop over checkpoint.artifact_paths only
checks PurePosixPath(path_str).is_absolute(), which misses Windows-style
absolute/UNC paths; update the validation in the artifact path loop (the code
referencing checkpoint.artifact_paths and PurePosixPath) to also detect Windows
absolute paths (e.g., by checking PureWindowsPath(path_str).is_absolute() or
using a cross-platform is-absolute test that covers both POSIX and Windows/UNC
forms) and append the same error message when either check is true.
In
`@tests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_list.py`:
- Around line 51-52: The test class TestHandlerCheckpointList is missing the
required pytest marker; add the `@pytest.mark.unit` decorator above the class
definition to classify it as a unit test, and ensure pytest is imported in the
file (e.g., add "import pytest" if not present) so the marker resolves
correctly.
In
`@tests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_write.py`:
- Around line 56-57: Add the pytest unit marker to the test class declaration:
decorate the TestHandlerCheckpointWrite class with `@pytest.mark.unit` (import
pytest if not already imported) so the class reads with the pytest marker above
it; ensure the marker is applied directly to the TestHandlerCheckpointWrite
class definition.
Comment on lines
+163
to
+166
| # 3. Path normalization: no absolute paths in artifact_paths | ||
| for path_str in checkpoint.artifact_paths: | ||
| if PurePosixPath(path_str).is_absolute(): | ||
| errors.append(f"Absolute artifact path forbidden: {path_str}") |
There was a problem hiding this comment.
Reject Windows absolute paths as well as POSIX.
Line 163-166 only use PurePosixPath.is_absolute(), which won’t flag Windows-style absolute paths like C:\... or UNC paths. This can bypass the “no absolute paths” guarantee.
🛡️ Suggested hardening
-from pathlib import PurePosixPath
+from pathlib import PurePosixPath, PureWindowsPath
@@
- for path_str in checkpoint.artifact_paths:
- if PurePosixPath(path_str).is_absolute():
+ for path_str in checkpoint.artifact_paths:
+ if (
+ PurePosixPath(path_str).is_absolute()
+ or PureWindowsPath(path_str).is_absolute()
+ ):
errors.append(f"Absolute artifact path forbidden: {path_str}")📝 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.
Suggested change
| # 3. Path normalization: no absolute paths in artifact_paths | |
| for path_str in checkpoint.artifact_paths: | |
| if PurePosixPath(path_str).is_absolute(): | |
| errors.append(f"Absolute artifact path forbidden: {path_str}") | |
| from pathlib import PurePosixPath, PureWindowsPath | |
| # 3. Path normalization: no absolute paths in artifact_paths | |
| for path_str in checkpoint.artifact_paths: | |
| if ( | |
| PurePosixPath(path_str).is_absolute() | |
| or PureWindowsPath(path_str).is_absolute() | |
| ): | |
| errors.append(f"Absolute artifact path forbidden: {path_str}") |
🤖 Prompt for AI Agents
In
`@src/omnibase_infra/nodes/node_checkpoint_validate_compute/handlers/handler_checkpoint_validate.py`
around lines 163 - 166, The current loop over checkpoint.artifact_paths only
checks PurePosixPath(path_str).is_absolute(), which misses Windows-style
absolute/UNC paths; update the validation in the artifact path loop (the code
referencing checkpoint.artifact_paths and PurePosixPath) to also detect Windows
absolute paths (e.g., by checking PureWindowsPath(path_str).is_absolute() or
using a cross-platform is-absolute test that covers both POSIX and Windows/UNC
forms) and append the same error message when either check is true.
Comment on lines
+51
to
+52
| class TestHandlerCheckpointList: | ||
| """Tests for HandlerCheckpointList.""" |
There was a problem hiding this comment.
Add @pytest.mark.unit marker to the test class.
The test class is missing the required pytest marker for test classification.
🔧 Proposed fix
+@pytest.mark.unit
class TestHandlerCheckpointList:
"""Tests for HandlerCheckpointList."""As per coding guidelines, Use pytest markers: @pytest.mark.unit, @pytest.mark.integration, @pytest.mark.slow, @pytest.mark.chaos, @pytest.mark.serial for test classification.
📝 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.
Suggested change
| class TestHandlerCheckpointList: | |
| """Tests for HandlerCheckpointList.""" | |
| `@pytest.mark.unit` | |
| class TestHandlerCheckpointList: | |
| """Tests for HandlerCheckpointList.""" |
🤖 Prompt for AI Agents
In `@tests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_list.py`
around lines 51 - 52, The test class TestHandlerCheckpointList is missing the
required pytest marker; add the `@pytest.mark.unit` decorator above the class
definition to classify it as a unit test, and ensure pytest is imported in the
file (e.g., add "import pytest" if not present) so the marker resolves
correctly.
Comment on lines
+56
to
+57
| class TestHandlerCheckpointWrite: | ||
| """Tests for HandlerCheckpointWrite.""" |
There was a problem hiding this comment.
Add @pytest.mark.unit marker to the test class.
The test class is missing the required pytest marker for test classification.
🔧 Proposed fix
+@pytest.mark.unit
class TestHandlerCheckpointWrite:
"""Tests for HandlerCheckpointWrite."""As per coding guidelines, Use pytest markers: @pytest.mark.unit, @pytest.mark.integration, @pytest.mark.slow, @pytest.mark.chaos, @pytest.mark.serial for test classification.
🤖 Prompt for AI Agents
In
`@tests/unit/nodes/test_node_checkpoint_effect/test_handler_checkpoint_write.py`
around lines 56 - 57, Add the pytest unit marker to the test class declaration:
decorate the TestHandlerCheckpointWrite class with `@pytest.mark.unit` (import
pytest if not already imported) so the class reads with the pytest marker above
it; ensure the marker is applied directly to the TestHandlerCheckpointWrite
class definition.
…dators, [minor] assert→raise, uuid4 fallback, glob precision Major: - Extract _attempt_number to shared util_checkpoint_sorting.py (was cross-module private import) - Add field_validator on repo_commit_map to enforce hex SHA format at model level - Add field_validator on artifact_paths to reject absolute paths and traversal at model level Minor: - Replace bare assert with explicit RuntimeError raise in 3 effect handlers - Add os.link() same-filesystem constraint comment - Document attempt_number return-0 behavior for non-matching files - Use uuid4() fallback instead of zero-UUID in validate handler - Tighten glob pattern to [0-9]* for attempt-number precision - Update tests to use model_construct() for intentionally-invalid data
…ove unused filename property - test_phase_payload_mismatch_is_error now creates a real mismatch (header=IMPLEMENT, payload.phase=local_review) and asserts Phase mismatch error - Remove unused EnumCheckpointPhase.filename property (misleading, no attempt suffix)
… counts Union threshold: 119 + 2 (OMN-2117) + 2 (OMN-2143) = 123
… counts (121→124)
jonahgabriel
deleted the
jonah/omn-2143-omnibase_infra-p4-checkpoint-nodes-checkpoint-effect
branch
February 11, 2026 22:16
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
ModelCheckpointwith discriminated union of 5 per-phase payloads (implement,local_review,create_pr,pr_release_ready,ready_for_merge)phase_numberandfilenameproperties for canonical checkpoint file namingCheckpoint storage layout
Key design decisions
_a2.yaml,_a3.yaml) — existing checkpoints are never overwrittenmodel_validatoronModelCheckpointenforces that the headerphasematches the discriminatedphase_payload.phaseFiles (39 changed, +2685)
enum_checkpoint_phase.py+__init__.pyexportmodels/checkpoint/infra_validators.py(union count bump),validation_exemptions.yamlTest plan
Ticket: OMN-2143
Summary by CodeRabbit
New Features
Chores
Tests