core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(workbench): persist construction evidence in trace journal

Browse source
This commit is contained in:
Shay 2026-06-23 13:40:51 -07:00
parent d7a809e82a
commit fbadac526f
5 changed files with 322 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

140
tests/test_workbench_construction_evidence.py
View file

@ -96,3 +96,143 @@ def test_existing_construction_evidence_instance_round_trips() -> None:
construction_evidence: ConstructionEvidence
assert construction_evidence_from_journal_entry(EntryWithPayload(11, existing)) is existing
def test_dict_deserialization_success() -> None:
payload = {
"schema_version": "construction_evidence_v1",
"turn_id": 12,
"status": "recorded",
"missing_reason": None,
"problem_text": "Lena has 3 red marbles.",
"proposals": [
{
"family_id": "fraction_decrease",
"relation_type": "decrease",
"candidate_organ": "assess_fraction_decrease",
"status": "proposed",
"evidence_spans": [{"start": 9, "end": 10, "text": "3"}],
"role_obligations": [
{"role": "quantity", "required": True, "description": "amount"}
],
"diagnostic_only": True,
"serving_allowed": False,
}
],
"mentions": [
{
"mention_id": "m1",
"kind": "quantity",
"surface": "3",
"span": {"start": 9, "end": 10, "text": "3"},
"fact_id": "f1",
}
],
"bindings": [
{
"binding_type": "mention_binding",
"source_mention_id": "m1",
"target_mention_id": "m1",
"evidence_spans": [{"start": 9, "end": 10, "text": "3"}],
}
],
"bound_relations": [
{
"relation_type": "decrease",
"roles": [
{
"role": "quantity",
"target_id": "m1",
"evidence_spans": [{"start": 9, "end": 10, "text": "3"}],
}
],
"evidence_spans": [{"start": 9, "end": 10, "text": "3"}],
}
],
"contract_assessments": [
{
"candidate_organ": "assess_fraction_decrease",
"family_id": "fraction_decrease",
"missing_bindings": [],
"unresolved_hazards": [],
"runnable": True,
"explanation": "good to go",
"evidence_spans": [{"start": 9, "end": 10, "text": "3"}],
}
],
"diagnostic_only": True,
"serving_allowed": False,
}
@dataclass(frozen=True, slots=True)
class EntryWithDict:
turn_id: int
construction_evidence: dict
evidence = construction_evidence_from_journal_entry(EntryWithDict(12, payload))
assert evidence.status == "recorded"
assert evidence.turn_id == 12
assert evidence.problem_text == "Lena has 3 red marbles."
assert len(evidence.proposals) == 1
assert evidence.proposals[0].family_id == "fraction_decrease"
assert len(evidence.proposals[0].evidence_spans) == 1
assert evidence.proposals[0].evidence_spans[0].text == "3"
assert len(evidence.mentions) == 1
assert evidence.mentions[0].mention_id == "m1"
assert evidence.mentions[0].span.start == 9
assert len(evidence.bindings) == 1
assert evidence.bindings[0].binding_type == "mention_binding"
assert len(evidence.bound_relations) == 1
assert evidence.bound_relations[0].roles[0].target_id == "m1"
assert len(evidence.contract_assessments) == 1
assert evidence.contract_assessments[0].runnable is True
def test_dict_deserialization_span_mismatch_refuses() -> None:
payload = {
"schema_version": "construction_evidence_v1",
"turn_id": 13,
"status": "recorded",
"missing_reason": None,
"problem_text": "Lena has 3 red marbles.",
"mentions": [
{
"mention_id": "m1",
"kind": "quantity",
"surface": "3",
"span": {"start": 9, "end": 10, "text": "wrong"}, # text mismatch!
"fact_id": "f1",
}
],
}
@dataclass(frozen=True, slots=True)
class EntryWithDict:
turn_id: int
construction_evidence: dict
evidence = construction_evidence_from_journal_entry(EntryWithDict(13, payload))
assert evidence.status == "missing_evidence"
assert evidence.turn_id == 13
assert "exact span validation failed" in evidence.missing_reason
def test_dict_deserialization_bad_schema_fails() -> None:
payload = {
"schema_version": "wrong_version_v2",
"turn_id": 14,
"status": "recorded",
"problem_text": "hello",
}
@dataclass(frozen=True, slots=True)
class EntryWithDict:
turn_id: int
construction_evidence: dict
evidence = construction_evidence_from_journal_entry(EntryWithDict(14, payload))
assert evidence.status == "missing_evidence"
assert "unsupported schema version" in evidence.missing_reason

31
tests/test_workbench_journal.py
View file

@ -10,6 +10,7 @@ from workbench import api as workbench_api
from workbench import readers
from workbench.api import MAX_CHAT_PROMPT_CHARS, WorkbenchApi
from workbench.journal import TurnJournal, TurnJournalEntry
from workbench.construction_evidence import ConstructionEvidence
from workbench.schemas import (
ChatTurnResult,
CognitivePipelineEdge,
@ -421,3 +422,33 @@ def test_unknown_turn_returns_404(tmp_path: Path) -> None:
assert missing.status == 404
assert invalid.status == 404
def test_trace_construction_endpoint_delivers_persisted_evidence(
tmp_path: Path,
) -> None:
journal = TurnJournal(tmp_path / "workbench_data")
evidence_payload = ConstructionEvidence(
schema_version="construction_evidence_v1",
turn_id=1,
status="recorded",
missing_reason=None,
problem_text="Lena has 3 red marbles.",
diagnostic_only=True,
serving_allowed=False,
)
entry = replace(_entry(1), construction_evidence=evidence_payload)
journal.append(entry)
api = WorkbenchApi(journal=journal)
response = _request(api, "GET", "/trace/1/construction")
assert response.status == 200
assert response.payload["ok"] is True
data = response.payload["data"]
assert data["schema_version"] == "construction_evidence_v1"
assert data["status"] == "recorded"
assert data["problem_text"] == "Lena has 3 red marbles."
assert data["diagnostic_only"] is True

144
workbench/construction_evidence.py
View file

@ -166,12 +166,148 @@ def construction_evidence_from_journal_entry(entry: Any) -> ConstructionEvidence
"construction evidence payload has unsupported shape",
)
# Future-compatible but conservative: do not attempt lossy coercion. The first
# persistence PR should replace this with explicit field-by-field projection and
# exact-span validation tests.
try:
schema_version = payload.get("schema_version")
if schema_version != "construction_evidence_v1":
return missing_construction_evidence(
turn_id, f"unsupported schema version: {schema_version}"
)
status = payload.get("status")
if status == "missing_evidence":
return missing_construction_evidence(
turn_id, payload.get("missing_reason") or CONSTRUCTION_EVIDENCE_ABSENT
)
if status != "recorded":
return missing_construction_evidence(
turn_id, f"unsupported status: {status}"
)
problem_text = payload.get("problem_text")
if problem_text is None:
return missing_construction_evidence(
turn_id, "recorded construction evidence missing problem_text"
)
if not isinstance(problem_text, str):
return missing_construction_evidence(
turn_id, "problem_text must be a string"
)
def parse_span(s: Any) -> SourceSpanView:
if not isinstance(s, dict):
raise TypeError("span must be a dict")
span = SourceSpanView(
start=int(s["start"]),
end=int(s["end"]),
text=str(s["text"]),
)
if not span_is_exact(problem_text, span):
raise ValueError(f"exact span validation failed: {span}")
return span
def parse_spans(lst: Any) -> list[SourceSpanView]:
if not isinstance(lst, list):
return []
return [parse_span(s) for s in lst]
proposals: list[ConstructionProposalView] = []
for prop in payload.get("proposals") or []:
role_obligations = [
RoleObligationView(
role=str(r["role"]),
required=bool(r["required"]),
description=str(r["description"]),
)
for r in prop.get("role_obligations") or []
]
proposals.append(
ConstructionProposalView(
family_id=str(prop["family_id"]),
relation_type=str(prop["relation_type"]),
candidate_organ=str(prop["candidate_organ"]),
status=prop["status"],
evidence_spans=parse_spans(prop.get("evidence_spans")),
role_obligations=role_obligations,
diagnostic_only=bool(prop.get("diagnostic_only", True)),
serving_allowed=bool(prop.get("serving_allowed", False)),
)
)
mentions: list[MentionView] = []
for m in payload.get("mentions") or []:
mentions.append(
MentionView(
mention_id=str(m["mention_id"]),
kind=str(m["kind"]),
surface=str(m["surface"]),
span=parse_span(m["span"]),
fact_id=m.get("fact_id"),
)
)
bindings: list[MentionBindingView] = []
for b in payload.get("bindings") or []:
bindings.append(
MentionBindingView(
binding_type=str(b["binding_type"]),
source_mention_id=str(b["source_mention_id"]),
target_mention_id=str(b["target_mention_id"]),
evidence_spans=parse_spans(b.get("evidence_spans")),
)
)
bound_relations: list[BoundRelationView] = []
for br in payload.get("bound_relations") or []:
roles = [
BoundRelationRoleView(
role=str(r["role"]),
target_id=str(r["target_id"]),
evidence_spans=parse_spans(r.get("evidence_spans")),
)
for r in br.get("roles") or []
]
bound_relations.append(
BoundRelationView(
relation_type=str(br["relation_type"]),
roles=roles,
evidence_spans=parse_spans(br.get("evidence_spans")),
)
)
contract_assessments: list[ContractAssessmentView] = []
for ca in payload.get("contract_assessments") or []:
contract_assessments.append(
ContractAssessmentView(
candidate_organ=str(ca["candidate_organ"]),
family_id=ca.get("family_id"),
missing_bindings=[str(mb) for mb in ca.get("missing_bindings") or []],
unresolved_hazards=[str(uh) for uh in ca.get("unresolved_hazards") or []],
runnable=bool(ca.get("runnable", False)),
explanation=str(ca.get("explanation", "")),
evidence_spans=parse_spans(ca.get("evidence_spans")),
)
)
return ConstructionEvidence(
schema_version="construction_evidence_v1",
turn_id=turn_id,
status="recorded",
missing_reason=None,
problem_text=problem_text,
proposals=proposals,
mentions=mentions,
bindings=bindings,
bound_relations=bound_relations,
contract_assessments=contract_assessments,
diagnostic_only=bool(payload.get("diagnostic_only", True)),
serving_allowed=bool(payload.get("serving_allowed", False)),
)
except (KeyError, TypeError, ValueError) as exc:
return missing_construction_evidence(
turn_id,
"construction evidence payload projection is not yet admitted",
f"construction evidence payload projection failed: {exc}",
)

3
workbench/journal.py
View file

@ -9,6 +9,7 @@ from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any
from workbench.construction_evidence import ConstructionEvidence
from workbench.schemas import (
ChatTurnResult,
CognitivePipelineRecord,
@ -58,6 +59,7 @@ class TurnJournalEntry:
leeway_evidence: dict[str, Any] | None = None
pipeline_record: CognitivePipelineRecord | dict[str, Any] | None = None
field_evidence: FieldEvidence | dict[str, Any] | None = None
construction_evidence: ConstructionEvidence | dict[str, Any] | None = None
trace_integrity: TraceIntegrity | None = None
journal_digest: str = ""
@ -99,6 +101,7 @@ class TurnJournalEntry:
leeway_evidence=to_data(result.leeway_evidence),
pipeline_record=to_data(result.pipeline_record),
field_evidence=to_data(result.field_evidence),
construction_evidence=to_data(result.construction_evidence),
trace_integrity=_trace_integrity_for_hash(result.trace_hash),
)

5
workbench/schemas.py
View file

@ -7,6 +7,9 @@ from datetime import datetime, timezone
from enum import Enum
from typing import Any, Literal
from workbench.construction_evidence import ConstructionEvidence
ErrorCode = Literal[
"bad_request",
@ -332,6 +335,7 @@ class ChatTurnResult:
leeway_evidence: LeewayEvidence | None = None
pipeline_record: CognitivePipelineRecord | None = None
field_evidence: FieldEvidence | None = None
construction_evidence: ConstructionEvidence | None = None
turn_id: int | None = None
@ -369,6 +373,7 @@ class TurnJournalEntrySchema:
leeway_evidence: LeewayEvidence | None = None
pipeline_record: CognitivePipelineRecord | None = None
field_evidence: FieldEvidence | None = None
construction_evidence: ConstructionEvidence | None = None
@dataclass(frozen=True, slots=True)