Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/research/NEXT_100_PR_MAP.md
Original file line number Diff line number Diff line change
Expand Up @@ -338,3 +338,4 @@ Machine-verifiable audit trail for claim downgrades, expert decisions, and exter
|---:|---|---|---|
| AB1 | Add claim strength downgrade record schema (CSD-). | Auditable trail for every claim downgrade; proof_ladder_steps_dropped auto-computed; is_retracted invariant; trigger_type controlled vocabulary (benchmark/reviewer/cheap_enemy/etc.); prevents silent claim drift upward after challenges. | B/C |
| AB2 | Add reviewer decision record schema (RDR-). | Machine-parseable expert review: 5 dimensions (novelty/controls/safety/synthesis/claim_scope); rating per dimension (acceptable/concerns_noted/requires_revision/not_assessed); n_blocking auto-computed; "approved" blocked when any required dimension unassessed or any dimension requires_revision. | B/C |
| AB3 | Add evidence gap notification schema (EGN-). | Structured record of what evidence is missing and how to close the gap; gap_type (9 types: missing_wet_lab/baseline/novelty/safety/reproducibility/reviewer/claim_mismatch/family_benchmark/adapter_baseline); closure_artifact_type (14 types); effort_estimate/priority/verdict; is_blocking flag; makes "needs more work" actionable. | C |
171 changes: 171 additions & 0 deletions src/openamp_foundry/evidence/evidence_gap_notification.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,171 @@
"""EGN- Evidence Gap Notification schema.

Machine-readable record of what evidence is missing and how to close
the gap. Makes "needs more work" outcomes actionable.
"""

from __future__ import annotations

from dataclasses import dataclass

VALID_EGN_GAP_TYPES: frozenset[str] = frozenset({
"missing_wet_lab_validation",
"missing_baseline_comparison",
"missing_novelty_evidence",
"missing_safety_clearance",
"missing_reproducibility_data",
"missing_reviewer_assessment",
"claim_strength_mismatch",
"missing_family_benchmark",
"missing_adapter_baseline",
})

VALID_EGN_CLOSURE_ARTIFACT_TYPES: frozenset[str] = frozenset({
"WHR", "CBR", "SEG", "CFC", "FNR", "PSC", "DCR", "RMC",
"RDR", "CSD", "FBH", "CBF", "BXR", "ARG",
})

VALID_EGN_EFFORT_ESTIMATES: frozenset[str] = frozenset({
"hours",
"days",
"weeks",
"months",
"unknown",
})

VALID_EGN_PRIORITIES: frozenset[str] = frozenset({
"critical", "high", "medium", "low",
})

VALID_EGN_VERDICTS: frozenset[str] = frozenset({
"actionable",
"blocked_on_resources",
"under_investigation",
"accepted_limitation",
})


@dataclass
class EvidenceGapNotification:
egn_id: str
pipeline_version: str
artifact_id: str
gap_type: str
gap_description: str
closure_artifact_type: str
closure_description: str
effort_estimate: str
priority: str
verdict: str
is_blocking: bool
limitations: list[str]
created_at: str
dry_lab_only: bool = True


def build_evidence_gap_notification(
*,
egn_id: str,
pipeline_version: str,
artifact_id: str,
gap_type: str,
gap_description: str,
closure_artifact_type: str,
closure_description: str,
effort_estimate: str,
priority: str,
verdict: str,
is_blocking: bool,
limitations: list[str],
created_at: str,
) -> EvidenceGapNotification:
egn = EvidenceGapNotification(
egn_id=egn_id,
pipeline_version=pipeline_version,
artifact_id=artifact_id,
gap_type=gap_type,
gap_description=gap_description,
closure_artifact_type=closure_artifact_type,
closure_description=closure_description,
effort_estimate=effort_estimate,
priority=priority,
verdict=verdict,
is_blocking=is_blocking,
dry_lab_only=True,
limitations=limitations,
created_at=created_at,
)
validate_evidence_gap_notification(egn)
return egn


def validate_evidence_gap_notification(
egn: EvidenceGapNotification,
) -> None:
if not egn.egn_id.startswith("EGN-"):
raise ValueError(
f"egn_id must start with 'EGN-': {egn.egn_id!r}"
)
if not egn.pipeline_version:
raise ValueError("pipeline_version must be non-empty")
if not egn.artifact_id:
raise ValueError("artifact_id must be non-empty")
if not egn.gap_description:
raise ValueError("gap_description must be non-empty")
if len(egn.gap_description) > 400:
raise ValueError(
f"gap_description exceeds 400 chars: {len(egn.gap_description)}"
)
if egn.gap_type not in VALID_EGN_GAP_TYPES:
raise ValueError(
f"gap_type {egn.gap_type!r} not in VALID_EGN_GAP_TYPES"
)
if not egn.closure_description:
raise ValueError("closure_description must be non-empty")
if egn.closure_artifact_type not in VALID_EGN_CLOSURE_ARTIFACT_TYPES:
raise ValueError(
f"closure_artifact_type {egn.closure_artifact_type!r} "
f"not in VALID_EGN_CLOSURE_ARTIFACT_TYPES"
)
if egn.effort_estimate not in VALID_EGN_EFFORT_ESTIMATES:
raise ValueError(
f"effort_estimate {egn.effort_estimate!r} "
f"not in VALID_EGN_EFFORT_ESTIMATES"
)
if egn.priority not in VALID_EGN_PRIORITIES:
raise ValueError(
f"priority {egn.priority!r} not in VALID_EGN_PRIORITIES"
)
if egn.verdict not in VALID_EGN_VERDICTS:
raise ValueError(
f"verdict {egn.verdict!r} not in VALID_EGN_VERDICTS"
)
if egn.dry_lab_only is not True:
raise ValueError("dry_lab_only must be True")
if not egn.limitations:
raise ValueError("limitations must be non-empty")
if not egn.created_at:
raise ValueError("created_at must be non-empty")


def format_evidence_gap_notification(
egn: EvidenceGapNotification,
) -> str:
lines = [
f"Evidence Gap Notification — {egn.egn_id}",
f"Pipeline: {egn.pipeline_version}",
f"Artifact: {egn.artifact_id}",
f"Gap type: {egn.gap_type}",
f"Gap description: {egn.gap_description}",
f"Closure artifact type: {egn.closure_artifact_type}",
f"Closure description: {egn.closure_description}",
f"Effort estimate: {egn.effort_estimate}",
f"Priority: {egn.priority}",
f"Verdict: {egn.verdict}",
f"Is blocking: {egn.is_blocking}",
]
for lim in egn.limitations:
lines.append(f" Limitation: {lim}")
lines.append(f"Dry lab only: {egn.dry_lab_only}")
lines.append(f"Created: {egn.created_at}")
return "\n".join(lines)
Loading
Loading