fix(experiments): return a corpus commit that contains every reported run set and result

[morluto]

Severity

High provenance and API-correctness — ExperimentRunResult.corpus_commit_id can point to a commit created before the run sets returned in the same result were published

Validated against main at 0a60b288342cd96e2608302fa9fd2917e59d7e58.

Audit only; no implementation change is included.

Summary

ExperimentService.run() stores the GenerationPublisher result from variant publication in a local variable named published.

It then freezes/publishes one run set per treatment. Those operations advance corpus HEAD, but their publication commit IDs are not captured in published.

When no outcome row or automatic comparison is subsequently published—for example:

• a performance experiment has other than exactly two run sets;
• a measurement experiment uses an adapter other than pyperf;
• failed/unsupported treatments leave fewer than two run sets;

the method returns:

corpus_commit_id = published.commit.commit_id

where published still refers to the earlier variants commit.

The same ExperimentRunResult contains the newly created run_sets, but those rows do not exist in the reported commit.

Direct code evidence

After trials, variants are published:

published = await run_atomic_thread(
    lambda: self.publisher.publish_rows(
        {"variants": [...]},
        ...,
    )
)

Then run sets are created through:

run_sets = await run_atomic_thread(
    lambda: tuple(
        RunSetService(self.workspace).freeze(...)
        for name in plan.variants
        ...
    )
)

Each freeze() independently publishes a run_sets row and advances HEAD, but its publication result is not assigned to published.

Later branches are:

if outcome experiment:
    published = publish experiment_outcomes
elif len(run_sets) != 2:
    limitations.append(...)
elif measurement adapter != pyperf:
    limitations.append(...)
else:
    comparison = ComparisonService(...).record(...)

Finally:

result_commit_id = published.commit.commit_id
if comparison is not None:
    result_commit_id = comparison.materialized_commit_id or comparison.corpus_commit_id

flameox/src/flameox/application/experiments.py

Lines 760 to 920 in 0a60b28

	cell.factors
	for block in plan.blocks
	for cell in block.cells
	if cell.treatment == name
	),
	{
	plan.variant_parameter: next(
	value
	for value in plan.factors[plan.variant_parameter]
	if self._factor_label(cast(Scalar, value)) == name
	)
	},
	)
	variants.append(
	Variant(
	variant_id=digest_model(
	{
	"experiment_id": plan.experiment.experiment_id,
	"name": name,
	}
	),
	experiment_id=plan.experiment.experiment_id,
	name=name,
	source_state_id=run.source_state_id if run is not None else None,
	workload_instance_id=(run.workload_instance_id if run is not None else None),
	parameters=factor_values,
	environment_requirements={},
	)
	)
	published = await run_atomic_thread(
	lambda: self.publisher.publish_rows(
	{
	"variants": [self._variant_row(value) for value in variants],
	},
	publisher="flameox.experiments",
	publisher_version="1",
	input_run_ids=tuple(trial.run_id for trial in trials if trial.run_id is not None),
	)
	)
	run_sets = await run_atomic_thread(
	lambda: tuple(
	RunSetService(self.workspace).freeze(
	FreezeRunMembersRequest(
	members=tuple(
	_freeze_trial_member(trial)
	for trial in trials_by_variant[name]
	if trial.run_id is not None
	),
	selection={
	"experiment_id": plan.experiment.experiment_id,
	"variant": name,
	},
	)
	)
	for name in plan.variants
	if any(trial.run_id is not None for trial in trials_by_variant[name])
	)
	)
	completed += 1
	await report("Variants and frozen run sets published")
	comparison: ComparisonResult | None = None
	outcome_result: OutcomeExperimentResult | None = None
	limitations: list[str] = []
	if isinstance(config, _OutcomeExperimentConfig):
	outcome_result = self._outcome_result(plan, config, trials)
	published = await run_atomic_thread(
	lambda: self.publisher.publish_rows(
	{"experiment_outcomes": [self._outcome_row(outcome_result)]},
	publisher="flameox.experiments",
	publisher_version="1",
	input_run_ids=tuple(
	trial.run_id for trial in trials if trial.run_id is not None
	),
	)
	)
	limitations.extend(outcome_result.limitations)
	elif len(run_sets) != 2:
	limitations.append(
	"Automatic paired comparison currently requires exactly two variants."
	)
	elif plan.metric_source == "measurement" and plan.adapter != "pyperf":
	limitations.append(
	"Automatic experiment comparison currently requires pyperf measurements."
	)
	else:
	comparison = await run_atomic_thread(
	lambda: ComparisonService(self.workspace).record(
	parse_compare_run_sets_request(
	{
	"baseline_run_set_id": run_sets[0].run_set_id,
	"candidate_run_set_id": run_sets[1].run_set_id,
	"experiment_id": plan.experiment.experiment_id,
	"metric": plan.experiment.primary_metric,
	"unit": ("bytes" if plan.metric_source == "runtime_resource" else "ns"),
	"metric_source": plan.metric_source,
	"polarity": plan.experiment.polarity,
	"practical_threshold": plan.experiment.practical_threshold,
	"confidence_level": plan.experiment.confidence_level,
	"random_seed": plan.experiment.random_seed,
	}
	)
	)
	)
	result_commit_id = published.commit.commit_id
	if comparison is not None:
	result_commit_id = comparison.materialized_commit_id or comparison.corpus_commit_id
	completed += 1
	await report("Experiment comparison and result complete")
	return ExperimentRunResult(
	experiment=plan.experiment,
	variants=tuple(variants),
	trials=tuple(trials),
	run_sets=run_sets,
	comparison=comparison,
	outcome=outcome_result,
	corpus_commit_id=result_commit_id,
	limitations=tuple(limitations),
	)
	async def _publish_unattempted(
	self,
	plan: ExperimentPlan,
	schedule: tuple[tuple[ExperimentBlock, int, ExperimentCell], ...],
	) -> None:
	for block, order, cell in schedule:
	trial = self._make_trial(
	plan=plan,
	cell=cell,
	run=None,
	block_id=block.block_id,
	order=order,
	outcome=TrialOutcome.UNATTEMPTED,
	failure_class="unattempted",
	)
	await run_atomic_thread(partial(self._publish_trial, trial))
	def _validate_plan(self, plan: ExperimentPlan) -> ExperimentConfig:
	if plan.workspace_id != self.workspace.identity.workspace_id:
	raise DomainError(ErrorCode.INVALID_CAPTURE_PLAN, "Workspace changed.")
	project = self.workloads.load()
	config = project.experiments[plan.experiment_name]
	if digest_model(config.model_dump(mode="json")) != plan.experiment_config_digest:
	raise DomainError(
	ErrorCode.INVALID_CAPTURE_PLAN,
	"Experiment definition changed after planning.",
	)
	definition = self.workloads.definition(config.workload)
	if definition.workload_definition_id != plan.experiment.workload_definition_id:
	raise DomainError(
	ErrorCode.INVALID_CAPTURE_PLAN,
	"Workload definition changed after experiment planning.",
	)
	return config
	def _materialize_combinations(
	self,
	config: ExperimentConfig,
	workload_parameters: dict[str, tuple[Scalar, ...]],
	) -> tuple[str, dict[str, tuple[Scalar, ...]], tuple[dict[str, Scalar], ...]]:
	if isinstance(config, _FactorExperimentConfig):
	treatment_factor = config.treatment_factor

Only the outcome/comparison paths update the result snapshot after run-set publication.

Deterministic three-variant proof

Run a successful pyperf performance experiment with treatments:

baseline
candidate_a
candidate_b

All three produce runs and trials.

The durable sequence is:

Ctrial...   trial publications
Cv          variants publication; local variable `published = Cv`
Cr1         baseline run-set publication
Cr2         candidate_a run-set publication
Cr3         candidate_b run-set publication

Now:

len(run_sets) == 3

so automatic paired comparison is skipped and only a limitation is appended.

No later publication occurs. The method returns:

result.corpus_commit_id = Cv
result.run_sets = (R1, R2, R3)
current HEAD = Cr3

At snapshot Cv:

SELECT * FROM run_sets WHERE run_set_id IN (R1,R2,R3)

returns zero rows.

Thus the result's declared snapshot cannot reproduce its own run-set fields.

Deterministic non-pyperf proof

Use two successful variants with a measurement-producing adapter other than pyperf.

The run sets publish at Cr1/Cr2, then:

elif plan.metric_source == "measurement" and plan.adapter != "pyperf":
    limitations.append(
        "Automatic experiment comparison currently requires pyperf measurements."
    )

Again no later publish updates published, so the result points to Cv rather than Cr2.

Partial-treatment proof

For a performance experiment where only one treatment has any run-bearing trial:

run_sets = (Rbaseline,)
len(run_sets) != 2

The returned commit predates even that one run set.

Why returning the earlier commit is not a harmless historical snapshot

A result snapshot is expected to identify the durable evidence represented by the response. Here it omits objects explicitly returned as completed outputs.

This affects:

• MCP resource links and follow-up lookups;
• reproducibility/audit logs;
• GC retention reasoning;
• summary/materialization provenance;
• clients that pin corpus_commit_id before navigating run_sets;
• retry/idempotency checks.

The result gives no second field naming the actual materialization commit for the run sets.

Additional consistency problem

Each run set itself pins a different pre-publication corpus commit because they are frozen sequentially (#210). Even using the latest run-set publication commit would not define one common analysis snapshot. The experiment needs an explicit terminal materialization/protocol that records all output identities together.

Violated invariant

For any structured result with corpus_commit_id=C:

all durable evidence objects returned by the result must be visible and verifiable in C

If several historical snapshots are intentionally referenced, their exact roles must be separately named; a single field cannot silently refer only to an early subset.

Proposed direction

1. Publish a terminal experiment-execution record

After variants, trials, run sets, and optional outcome/comparison are complete, publish one terminal result row/record containing their exact IDs, statuses, limitations, and source snapshots.

Return that materialization commit.

2. Use one coherent run-set snapshot protocol

Compose with #210/#281 so all treatment run sets are frozen from one declared common snapshot and remain retained through publication.

3. Distinguish source and materialization commits

Where needed, expose typed fields such as:

protocol_commit_id
trial_population_commit_id
run_set_source_commit_id
result_materialized_commit_id
comparison_commit_id

Do not overload one corpus_commit_id.

4. Validate before returning

Open the reported commit and prove every returned variant/trial/run-set/outcome/comparison ID is reachable.

Acceptance criteria

• ExperimentRunResult.corpus_commit_id contains every returned durable result object.
• Three-variant experiments return a commit containing all three run-set rows.
• Non-pyperf/no-comparison branches still materialize a terminal experiment result.
• Partial-treatment branches include the run sets/trials they return.
• Source snapshots and result materialization snapshots are explicitly distinguished.
• Every returned ID is verified against the reported commit before response construction.
• Tests cover 1, 2, and 3+ variants, outcome experiments, unsupported auto-comparison adapters, partial failures, and comparison success.
• Current HEAD advancement after return cannot change what the result commit means.

Relationship to existing issues

• fix(comparisons): honor frozen run-set corpus snapshots instead of latest HEAD #210 requires run sets to share and honor a frozen evidence snapshot.
• fix(run sets): retain the frozen corpus snapshot until run-set publication commits #281 keeps that frozen snapshot alive through run-set publication.
• [Architecture] Eliminate run-manifest/evidence dual writes with a recoverable publication transaction #160 defines the multi-stage publication/outbox consistency protocol.
• fix(summaries): resolve every run, finding, and reference from one pinned evidence snapshot #264/fix(evidence lookup): require every returned object to belong to the reported corpus snapshot #267 enforce the same one-snapshot rule for summaries/lookups.
• [Architecture] Add an explicit migration contract for every persisted schema #157 owns result-contract migration.