PR 05: Submit and materialize samples

docs/architecture/04_persistence.md

@@ -70,6 +70,23 @@ rows and eagerly creates every statically-declared node and edge. Task
 payloads are attached as annotations in the core-reserved `"payload"`
 namespace.
 
+PR05 adds the public experiment submit path alongside that bridge:
+`Experiment.submit(...)` persists an experiment and candidate pool, records the
+sampler invocation, marks selected candidate rows, and creates one
+`SampleRecord` per selected sample. The materialization boundary is the authored
+`Sample`'s concrete `Task` objects: each task is persisted as
+`task.model_dump(mode="json")` into typed sample WAL rows and the
+`sample_graph_nodes.task_json` projection. Environment and experiment objects
+remain provenance rows; they are not serialized into the runtime replay
+contract. `workflow/started` may now carry only `sample_id` when the sample
+graph already exists, while the definition-backed initialization path remains as
+a temporary bridge.
+
+TODO(PR09): remove the definition-backed initialization bridge when runtime
+launch no longer reads `experiment_definitions`, `experiment_definition_tasks`,
+or definition worker/evaluator bindings. The remaining path should initialize
+from already-materialized sample graph/WAL rows only.
+
 ### Manager-spawned subtasks (dynamic graph growth)
 
 The graph is append-only at the row level: new nodes and edges can enter

ergon-dashboard/src/generated/events/schemas/DashboardWorkflowStartedEvent.schema.json

@@ -1092,8 +1092,16 @@
           "type": "string"
         },
         "definitionId": {
-          "title": "Definitionid",
-          "type": "string"
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "title": "Definitionid"
         },
         "name": {
           "title": "Name",
@@ -1272,7 +1280,6 @@
       },
       "required": [
         "id",
-        "definitionId",
         "name",
         "status"
       ],

ergon_core/ergon_core/core/application/events/runtime.py

@@ -7,7 +7,7 @@
 
 from __future__ import annotations
 
-from typing import ClassVar, Literal
+from typing import Any, ClassVar, Literal
 from uuid import UUID
 
 from ergon_core.core.application.events.base import InngestEventContract
@@ -31,15 +31,15 @@ class TaskReadyEvent(InngestEventContract):
     name: ClassVar[str] = "task/ready"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
     task_id: UUID
 
 
 class TaskStartedEvent(InngestEventContract):
     name: ClassVar[str] = "task/started"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
     task_id: UUID
     execution_id: UUID
 
@@ -58,7 +58,7 @@ class TaskCancelledEvent(InngestEventContract):
     name: ClassVar[str] = "task/cancelled"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
     task_id: UUID
     execution_id: UUID | None
     cause: CancelCause
@@ -70,7 +70,7 @@ class TaskCompletedEvent(InngestEventContract):
     name: ClassVar[str] = "task/completed"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
     task_id: UUID
     execution_id: UUID
     sandbox_id: str
@@ -80,7 +80,7 @@ class TaskFailedEvent(InngestEventContract):
     name: ClassVar[str] = "task/failed"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
     task_id: UUID
     execution_id: UUID
     error: str
@@ -91,19 +91,23 @@ class WorkflowStartedEvent(InngestEventContract):
     name: ClassVar[str] = "workflow/started"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
+
+    def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
+        kwargs.setdefault("exclude_none", True)
+        return super().model_dump(*args, **kwargs)
 
 
 class WorkflowCompletedEvent(InngestEventContract):
     name: ClassVar[str] = "workflow/completed"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
 
 
 class WorkflowFailedEvent(InngestEventContract):
     name: ClassVar[str] = "workflow/failed"
 
     sample_id: UUID
-    definition_id: UUID
+    definition_id: UUID | None = None
     error: str

ergon_core/ergon_core/core/application/experiments/candidate_pool.py

@@ -5,9 +5,10 @@
 
 from sqlmodel import Session
 
+from ergon_core.api.benchmark import Task
+from ergon_core.api.benchmark.task import EmptyTaskPayload
 from ergon_core.api.experiment.experiment import Experiment, ExperimentRef
 from ergon_core.api.experiment.sample import Sample
-from ergon_core.api.benchmark import Task
 from ergon_core.core.application.experiments.repository import (
     ExperimentRepository,
 )
@@ -126,5 +127,9 @@ async def _task_from_candidate_snapshot(task_json: object) -> Task:
     # Candidate-pool rows are not runtime materializations. We use the existing
     # `_type` dispatch path to rebuild object-bound config, then clear the
     # temporary id so materialization remains the only runtime-id boundary.
+    if isinstance(task.dependency_task_slugs, list):
+        task.dependency_task_slugs = tuple(task.dependency_task_slugs)
+    if isinstance(task.task_payload, dict) and not task.task_payload:
+        task.task_payload = EmptyTaskPayload()
     task._task_id = None
     return task

ergon_core/ergon_core/core/application/experiments/persistence.py

@@ -17,4 +17,9 @@ def __init__(self, session: Session) -> None:
         self._session = session
 
     async def persist_experiment(self, experiment: Experiment) -> ExperimentRef:
-        return persist_row_graph(session=self._session, experiment=experiment)
+        ref = experiment.persisted_ref()
+        if ref is not None:
+            return ref
+        ref = persist_row_graph(session=self._session, experiment=experiment)
+        experiment.mark_persisted(ref)
+        return ref

ergon_core/ergon_core/core/application/experiments/repository.py

@@ -61,30 +61,6 @@ def persist_experiment(self, experiment: Experiment) -> ExperimentRef:
             metadata=row.metadata_json,
         )
 
-    def record_sampler_invocation(
-        self,
-        *,
-        experiment_ref: ExperimentRef,
-        sampler_name: str,
-        requested_k: int,
-        candidate_pool_size: int,
-        selected_count: int = 0,
-        policy_version: int | None = None,
-        sampler_config: dict[str, JsonValue] | None = None,
-    ) -> ExperimentSamplerInvocationRow:
-        row = ExperimentSamplerInvocationRow(
-            experiment_id=experiment_ref.experiment_id,
-            sampler_name=sampler_name,
-            requested_k=requested_k,
-            candidate_pool_size=candidate_pool_size,
-            selected_count=selected_count,
-            policy_version=policy_version,
-            sampler_config_json=dict(sampler_config or {}),
-        )
-        self._session.add(row)
-        self._session.flush()
-        return row
-
     def pending_unselected_pool_entries(
         self,
         experiment_id: UUID,
@@ -172,12 +148,15 @@ def record_sampler_invocation(
     policy_version: int | None = None,
     sampler_config: dict[str, JsonValue] | None = None,
 ) -> ExperimentSamplerInvocationRow:
-    return ExperimentRepository(session).record_sampler_invocation(
-        experiment_ref=experiment_ref,
+    row = ExperimentSamplerInvocationRow(
+        experiment_id=experiment_ref.experiment_id,
         sampler_name=sampler_name,
         requested_k=requested_k,
         candidate_pool_size=candidate_pool_size,
         selected_count=selected_count,
         policy_version=policy_version,
-        sampler_config=sampler_config,
+        sampler_config_json=dict(sampler_config or {}),
     )
+    session.add(row)
+    session.flush()
+    return row