diff --git a/README.md b/README.md index d7b49db..b355467 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,7 @@ Documentation for the whole stack lives at | `ActionResult` | Execution outcome with error taxonomy + state delta | | `Episode` / `Step` | Complete task trajectory (observation → action → result) | | `FailureRecord` | Classified failure for dataset pipelines | +| `ControlOverlayFrameV1` / `ControlOverlayTimelineV1` | PHI-safe execution overlay state bound to exact evidence media | ## Quick start @@ -140,6 +141,23 @@ schema = ComputerState.model_json_schema() print(json.dumps(schema, indent=2)) ``` +The same API exports the versioned cross-surface overlay contracts: + +```python +from openadapt_types import ControlOverlayFrameV1, ControlOverlayTimelineV1 + +frame_schema = ControlOverlayFrameV1.model_json_schema() +timeline_schema = ControlOverlayTimelineV1.model_json_schema() +``` + +The same schemas ship as `openadapt_types/schemas/control-overlay-frame-v1.json` +and `control-overlay-timeline-v1.json` for TypeScript, Rust, and other consumers. + +Overlay schemas reject unknown fields and contain only closed presentation +labels and canonical statuses. Screenshots, action targets, typed values, +identities, URLs, logs, report bodies, and user-authored workflow names remain +outside this public presentation contract. + ## Design principles - **Pydantic v2**: runtime validation, JSON Schema export, fast serialization diff --git a/openadapt_types/__init__.py b/openadapt_types/__init__.py index 0afb470..ecdd12d 100644 --- a/openadapt_types/__init__.py +++ b/openadapt_types/__init__.py @@ -39,6 +39,28 @@ ProcessInfo, UINode, ) +from openadapt_types.control_overlay import ( + CONTROL_OVERLAY_FRAME_SCHEMA, + CONTROL_OVERLAY_STATE_ID_COMPONENTS, + CONTROL_OVERLAY_STATUS_BY_PHASE, + CONTROL_OVERLAY_TERMINAL_PHASES, + CONTROL_OVERLAY_TIMELINE_SCHEMA, + CONTROL_OVERLAY_WORKFLOW_LABEL_BY_MODE, + ControlOverlayControlsV1, + ControlOverlayDataClassification, + ControlOverlayFrameV1, + ControlOverlayMode, + ControlOverlayPhase, + ControlOverlayProfile, + ControlOverlayStepV1, + ControlOverlayTimelineBindingV1, + ControlOverlayTimelineEventV1, + ControlOverlayTimelineV1, + ControlOverlayWorkflowLabel, + build_control_overlay_timeline, + control_overlay_state_id, + is_terminal_control_overlay_phase, +) from openadapt_types.episode import Episode, Step from openadapt_types.failure import FailureCategory, FailureRecord from openadapt_types.parsing import ( @@ -58,6 +80,27 @@ "ElementRole", "ProcessInfo", "UINode", + # control_overlay + "CONTROL_OVERLAY_FRAME_SCHEMA", + "CONTROL_OVERLAY_STATE_ID_COMPONENTS", + "CONTROL_OVERLAY_STATUS_BY_PHASE", + "CONTROL_OVERLAY_TERMINAL_PHASES", + "CONTROL_OVERLAY_TIMELINE_SCHEMA", + "CONTROL_OVERLAY_WORKFLOW_LABEL_BY_MODE", + "ControlOverlayControlsV1", + "ControlOverlayDataClassification", + "ControlOverlayFrameV1", + "ControlOverlayMode", + "ControlOverlayPhase", + "ControlOverlayProfile", + "ControlOverlayStepV1", + "ControlOverlayTimelineBindingV1", + "ControlOverlayTimelineEventV1", + "ControlOverlayTimelineV1", + "ControlOverlayWorkflowLabel", + "build_control_overlay_timeline", + "control_overlay_state_id", + "is_terminal_control_overlay_phase", # action "Action", "ActionResult", diff --git a/openadapt_types/control_overlay.py b/openadapt_types/control_overlay.py new file mode 100644 index 0000000..8eaa7b8 --- /dev/null +++ b/openadapt_types/control_overlay.py @@ -0,0 +1,395 @@ +"""Privacy-safe control-overlay contracts shared across OpenAdapt surfaces. + +The overlay is a presentation and control surface, not an evidence payload or +an execution authority. These models deliberately have no free-form runtime +text, screenshots, targets, typed values, identities, URLs, logs, or report +bodies. Producers keep those values inside their declared evidence boundary. +""" + +from __future__ import annotations + +from enum import Enum +from math import isfinite +from types import MappingProxyType +from typing import Literal, Mapping, Sequence + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + StrictBool, + StrictFloat, + StrictInt, + StrictStr, + model_validator, +) + +CONTROL_OVERLAY_FRAME_SCHEMA = "openadapt.control-overlay-frame/v1" +CONTROL_OVERLAY_TIMELINE_SCHEMA = "openadapt.control-overlay-timeline/v1" + + +class ControlOverlayPhase(str, Enum): + """Canonical cross-surface execution and terminal states.""" + + IDLE = "idle" + OBSERVING = "observing" + RECORDING = "recording" + EXECUTING = "executing" + PAUSING = "pausing" + PAUSED = "paused" + RESUMING = "resuming" + STOPPING = "stopping" + VERIFYING = "verifying" + VERIFIED = "verified" + COMPLETED_UNVERIFIED = "completed_unverified" + HALTED = "halted" + FAILED = "failed" + ROLLED_BACK = "rolled_back" + + +class ControlOverlayMode(str, Enum): + DEMONSTRATION = "demonstration" + REPLAY = "replay" + GOVERNED = "governed" + MANAGED = "managed" + + +class ControlOverlayProfile(str, Enum): + DEMO = "demo" + STANDARD = "standard" + REGULATED = "regulated" + + +class ControlOverlayDataClassification(str, Enum): + """The only classifications allowed to leave a private evidence boundary.""" + + SYNTHETIC = "synthetic" + SANITIZED_PUBLIC = "sanitized_public" + + +class ControlOverlayWorkflowLabel(str, Enum): + """Closed presentation labels; user-authored workflow names are excluded.""" + + DEMONSTRATION = "Workflow demonstration" + REPLAY = "Workflow replay" + GOVERNED = "Governed workflow" + MANAGED = "Managed workflow" + + +CONTROL_OVERLAY_STATUS_BY_PHASE: Mapping[ControlOverlayPhase, str] = MappingProxyType( + { + ControlOverlayPhase.IDLE: "Ready", + ControlOverlayPhase.OBSERVING: "Observing the application", + ControlOverlayPhase.RECORDING: "Watching your demonstration", + ControlOverlayPhase.EXECUTING: "Executing with verification gates", + ControlOverlayPhase.PAUSING: "Pausing at a safe boundary", + ControlOverlayPhase.PAUSED: "Execution paused", + ControlOverlayPhase.RESUMING: "Resuming at a safe boundary", + ControlOverlayPhase.STOPPING: "Stopping at a safe boundary", + ControlOverlayPhase.VERIFYING: "Verifying the intended result", + ControlOverlayPhase.VERIFIED: "Outcome verified", + ControlOverlayPhase.COMPLETED_UNVERIFIED: ( + "Completed without sufficient verification" + ), + ControlOverlayPhase.HALTED: "Halted instead of guessing", + ControlOverlayPhase.FAILED: "Execution failed", + ControlOverlayPhase.ROLLED_BACK: "Compensating action completed", + } +) + +CONTROL_OVERLAY_WORKFLOW_LABEL_BY_MODE: Mapping[ + ControlOverlayMode, ControlOverlayWorkflowLabel +] = MappingProxyType( + { + ControlOverlayMode.DEMONSTRATION: ControlOverlayWorkflowLabel.DEMONSTRATION, + ControlOverlayMode.REPLAY: ControlOverlayWorkflowLabel.REPLAY, + ControlOverlayMode.GOVERNED: ControlOverlayWorkflowLabel.GOVERNED, + ControlOverlayMode.MANAGED: ControlOverlayWorkflowLabel.MANAGED, + } +) + +CONTROL_OVERLAY_TERMINAL_PHASES = frozenset( + { + ControlOverlayPhase.VERIFIED, + ControlOverlayPhase.COMPLETED_UNVERIFIED, + ControlOverlayPhase.HALTED, + ControlOverlayPhase.FAILED, + ControlOverlayPhase.ROLLED_BACK, + } +) + +CONTROL_OVERLAY_STATE_ID_COMPONENTS = ( + "visibility", + "phase", + "mode", + "profile", + "step.current", + "step.total", + "controls.pause", + "controls.resume", + "controls.stop", +) + +_CONTROL_OVERLAY_FRAME_JSON_SCHEMA_EXTRA = { + "x-openadapt-status-by-phase": { + phase.value: status for phase, status in CONTROL_OVERLAY_STATUS_BY_PHASE.items() + }, + "x-openadapt-workflow-label-by-mode": { + mode.value: label.value + for mode, label in CONTROL_OVERLAY_WORKFLOW_LABEL_BY_MODE.items() + }, + "x-openadapt-terminal-phases": sorted( + phase.value for phase in CONTROL_OVERLAY_TERMINAL_PHASES + ), + "x-openadapt-state-id-components": list(CONTROL_OVERLAY_STATE_ID_COMPONENTS), +} + + +class _StrictContract(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class ControlOverlayStepV1(_StrictContract): + current: StrictInt | None = Field(default=None, ge=1) + total: StrictInt | None = Field(default=None, ge=1) + + @model_validator(mode="after") + def _validate_progress(self) -> "ControlOverlayStepV1": + if (self.current is None) != (self.total is None): + raise ValueError("step.current and step.total must both be set or null") + if ( + self.current is not None + and self.total is not None + and self.current > self.total + ): + raise ValueError("step.current cannot exceed step.total") + return self + + +class ControlOverlayControlsV1(_StrictContract): + pause: StrictBool = False + resume: StrictBool = False + stop: StrictBool = False + + +def control_overlay_state_id( + *, + visible: bool, + phase: ControlOverlayPhase, + mode: ControlOverlayMode, + profile: ControlOverlayProfile | None, + step: ControlOverlayStepV1, + controls: ControlOverlayControlsV1, +) -> str: + """Build the semantic identity used for equivalent overlay frames.""" + + return ":".join( + ( + "visible" if visible else "hidden", + phase.value, + mode.value, + profile.value if profile is not None else "no-profile", + str(step.current) if step.current is not None else "no-step", + str(step.total) if step.total is not None else "no-total", + "pause" if controls.pause else "no-pause", + "resume" if controls.resume else "no-resume", + "stop" if controls.stop else "no-stop", + ) + ) + + +class ControlOverlayFrameV1(_StrictContract): + """A deterministic frame safe for public presentation and composition.""" + + model_config = ConfigDict( + json_schema_extra=_CONTROL_OVERLAY_FRAME_JSON_SCHEMA_EXTRA + ) + + schema_version: Literal["openadapt.control-overlay-frame/v1"] = ( + CONTROL_OVERLAY_FRAME_SCHEMA + ) + state_id: StrictStr = Field(min_length=1, max_length=256) + event_sequence: StrictInt = Field(ge=0) + observed_at_unix_ms: StrictInt = Field(ge=0) + observed_at_monotonic_ms: StrictFloat = Field(ge=0) + visible: StrictBool + phase: ControlOverlayPhase + workflow_label: ControlOverlayWorkflowLabel + mode: ControlOverlayMode + profile: ControlOverlayProfile | None = None + step: ControlOverlayStepV1 + controls: ControlOverlayControlsV1 + status: StrictStr = Field(min_length=1, max_length=64) + presentation: Literal[True] = True + + @model_validator(mode="after") + def _validate_derived_fields(self) -> "ControlOverlayFrameV1": + if not isfinite(self.observed_at_monotonic_ms): + raise ValueError("observed_at_monotonic_ms must be finite") + expected_label = CONTROL_OVERLAY_WORKFLOW_LABEL_BY_MODE[self.mode] + if self.workflow_label != expected_label: + raise ValueError("workflow_label does not match mode") + expected_status = CONTROL_OVERLAY_STATUS_BY_PHASE[self.phase] + if self.status != expected_status: + raise ValueError("status does not match phase") + expected_state_id = control_overlay_state_id( + visible=self.visible, + phase=self.phase, + mode=self.mode, + profile=self.profile, + step=self.step, + controls=self.controls, + ) + if self.state_id != expected_state_id: + raise ValueError("state_id does not match the semantic frame state") + return self + + @classmethod + def build( + cls, + *, + event_sequence: int, + observed_at_unix_ms: int, + observed_at_monotonic_ms: float, + visible: bool, + phase: ControlOverlayPhase, + mode: ControlOverlayMode, + profile: ControlOverlayProfile | None = None, + current_step: int | None = None, + total_steps: int | None = None, + pause: bool = False, + resume: bool = False, + stop: bool = False, + ) -> "ControlOverlayFrameV1": + """Build a frame while deriving every presentation string and state ID.""" + + step = ControlOverlayStepV1(current=current_step, total=total_steps) + controls = ControlOverlayControlsV1( + pause=pause, + resume=resume, + stop=stop, + ) + return cls( + state_id=control_overlay_state_id( + visible=visible, + phase=phase, + mode=mode, + profile=profile, + step=step, + controls=controls, + ), + event_sequence=event_sequence, + observed_at_unix_ms=observed_at_unix_ms, + observed_at_monotonic_ms=observed_at_monotonic_ms, + visible=visible, + phase=phase, + workflow_label=CONTROL_OVERLAY_WORKFLOW_LABEL_BY_MODE[mode], + mode=mode, + profile=profile, + step=step, + controls=controls, + status=CONTROL_OVERLAY_STATUS_BY_PHASE[phase], + ) + + +class ControlOverlayTimelineEventV1(_StrictContract): + at_ms: StrictInt = Field(ge=0) + frame: ControlOverlayFrameV1 + + +class ControlOverlayTimelineBindingV1(_StrictContract): + evidence_pack_id: StrictStr = Field( + min_length=1, + max_length=96, + pattern=r"^[a-z0-9][a-z0-9._-]{0,95}$", + ) + media_sha256: StrictStr = Field(pattern=r"^[a-f0-9]{64}$") + + +class ControlOverlayTimelineV1(_StrictContract): + """Presentation frames bound to one exact evidence pack and media clip.""" + + schema_version: Literal["openadapt.control-overlay-timeline/v1"] = ( + CONTROL_OVERLAY_TIMELINE_SCHEMA + ) + data_classification: ControlOverlayDataClassification + evidence_pack_id: StrictStr = Field( + min_length=1, + max_length=96, + pattern=r"^[a-z0-9][a-z0-9._-]{0,95}$", + ) + media_sha256: StrictStr = Field(pattern=r"^[a-f0-9]{64}$") + duration_ms: StrictInt = Field(gt=0) + events: tuple[ControlOverlayTimelineEventV1, ...] = Field(min_length=1) + + @model_validator(mode="after") + def _validate_event_order(self) -> "ControlOverlayTimelineV1": + if self.events[0].at_ms != 0: + raise ValueError("timeline must begin at 0 ms") + previous_at = -1 + previous_sequence = -1 + previous_monotonic = -1.0 + for event in self.events: + if event.at_ms <= previous_at or event.at_ms > self.duration_ms: + raise ValueError("timeline media offsets must increase within duration") + if event.frame.event_sequence <= previous_sequence: + raise ValueError( + "timeline event_sequence values must strictly increase" + ) + if event.frame.observed_at_monotonic_ms < previous_monotonic: + raise ValueError("timeline monotonic timestamps cannot go backwards") + previous_at = event.at_ms + previous_sequence = event.frame.event_sequence + previous_monotonic = event.frame.observed_at_monotonic_ms + return self + + def assert_binding(self, binding: ControlOverlayTimelineBindingV1) -> None: + """Refuse composition with a different pack or exact media digest.""" + + if self.evidence_pack_id != binding.evidence_pack_id: + raise ValueError("timeline belongs to a different evidence pack") + if self.media_sha256 != binding.media_sha256: + raise ValueError("timeline does not match the exact media digest") + + def frame_at(self, current_time_ms: int | float) -> ControlOverlayFrameV1: + """Return the latest frame at a bounded media time.""" + + if isinstance(current_time_ms, bool) or not isinstance( + current_time_ms, (int, float) + ): + raise TypeError("current_time_ms must be a number") + if not isfinite(current_time_ms): + raise ValueError("current_time_ms must be finite") + bounded_time = min(max(current_time_ms, 0), self.duration_ms) + low = 0 + high = len(self.events) - 1 + while low < high: + midpoint = (low + high + 1) // 2 + if self.events[midpoint].at_ms <= bounded_time: + low = midpoint + else: + high = midpoint - 1 + return self.events[low].frame + + +def build_control_overlay_timeline( + *, + data_classification: ControlOverlayDataClassification, + evidence_pack_id: str, + media_sha256: str, + duration_ms: int, + events: Sequence[ControlOverlayTimelineEventV1], +) -> ControlOverlayTimelineV1: + """Build and validate an immutable control-overlay timeline.""" + + return ControlOverlayTimelineV1( + data_classification=data_classification, + evidence_pack_id=evidence_pack_id, + media_sha256=media_sha256, + duration_ms=duration_ms, + events=tuple(events), + ) + + +def is_terminal_control_overlay_phase(phase: ControlOverlayPhase) -> bool: + return phase in CONTROL_OVERLAY_TERMINAL_PHASES diff --git a/openadapt_types/schemas/__init__.py b/openadapt_types/schemas/__init__.py new file mode 100644 index 0000000..a4f606e --- /dev/null +++ b/openadapt_types/schemas/__init__.py @@ -0,0 +1 @@ +"""Packaged language-agnostic schemas generated from canonical Pydantic types.""" diff --git a/openadapt_types/schemas/control-overlay-frame-v1.json b/openadapt_types/schemas/control-overlay-frame-v1.json new file mode 100644 index 0000000..8b56319 --- /dev/null +++ b/openadapt_types/schemas/control-overlay-frame-v1.json @@ -0,0 +1,238 @@ +{ + "$defs": { + "ControlOverlayControlsV1": { + "additionalProperties": false, + "properties": { + "pause": { + "default": false, + "title": "Pause", + "type": "boolean" + }, + "resume": { + "default": false, + "title": "Resume", + "type": "boolean" + }, + "stop": { + "default": false, + "title": "Stop", + "type": "boolean" + } + }, + "title": "ControlOverlayControlsV1", + "type": "object" + }, + "ControlOverlayMode": { + "enum": [ + "demonstration", + "replay", + "governed", + "managed" + ], + "title": "ControlOverlayMode", + "type": "string" + }, + "ControlOverlayPhase": { + "description": "Canonical cross-surface execution and terminal states.", + "enum": [ + "idle", + "observing", + "recording", + "executing", + "pausing", + "paused", + "resuming", + "stopping", + "verifying", + "verified", + "completed_unverified", + "halted", + "failed", + "rolled_back" + ], + "title": "ControlOverlayPhase", + "type": "string" + }, + "ControlOverlayProfile": { + "enum": [ + "demo", + "standard", + "regulated" + ], + "title": "ControlOverlayProfile", + "type": "string" + }, + "ControlOverlayStepV1": { + "additionalProperties": false, + "properties": { + "current": { + "anyOf": [ + { + "minimum": 1, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Current" + }, + "total": { + "anyOf": [ + { + "minimum": 1, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Total" + } + }, + "title": "ControlOverlayStepV1", + "type": "object" + }, + "ControlOverlayWorkflowLabel": { + "description": "Closed presentation labels; user-authored workflow names are excluded.", + "enum": [ + "Workflow demonstration", + "Workflow replay", + "Governed workflow", + "Managed workflow" + ], + "title": "ControlOverlayWorkflowLabel", + "type": "string" + } + }, + "additionalProperties": false, + "description": "A deterministic frame safe for public presentation and composition.", + "properties": { + "controls": { + "$ref": "#/$defs/ControlOverlayControlsV1" + }, + "event_sequence": { + "minimum": 0, + "title": "Event Sequence", + "type": "integer" + }, + "mode": { + "$ref": "#/$defs/ControlOverlayMode" + }, + "observed_at_monotonic_ms": { + "minimum": 0, + "title": "Observed At Monotonic Ms", + "type": "number" + }, + "observed_at_unix_ms": { + "minimum": 0, + "title": "Observed At Unix Ms", + "type": "integer" + }, + "phase": { + "$ref": "#/$defs/ControlOverlayPhase" + }, + "presentation": { + "const": true, + "default": true, + "title": "Presentation", + "type": "boolean" + }, + "profile": { + "anyOf": [ + { + "$ref": "#/$defs/ControlOverlayProfile" + }, + { + "type": "null" + } + ], + "default": null + }, + "schema_version": { + "const": "openadapt.control-overlay-frame/v1", + "default": "openadapt.control-overlay-frame/v1", + "title": "Schema Version", + "type": "string" + }, + "state_id": { + "maxLength": 256, + "minLength": 1, + "title": "State Id", + "type": "string" + }, + "status": { + "maxLength": 64, + "minLength": 1, + "title": "Status", + "type": "string" + }, + "step": { + "$ref": "#/$defs/ControlOverlayStepV1" + }, + "visible": { + "title": "Visible", + "type": "boolean" + }, + "workflow_label": { + "$ref": "#/$defs/ControlOverlayWorkflowLabel" + } + }, + "required": [ + "state_id", + "event_sequence", + "observed_at_unix_ms", + "observed_at_monotonic_ms", + "visible", + "phase", + "workflow_label", + "mode", + "step", + "controls", + "status" + ], + "title": "ControlOverlayFrameV1", + "type": "object", + "x-openadapt-state-id-components": [ + "visibility", + "phase", + "mode", + "profile", + "step.current", + "step.total", + "controls.pause", + "controls.resume", + "controls.stop" + ], + "x-openadapt-status-by-phase": { + "completed_unverified": "Completed without sufficient verification", + "executing": "Executing with verification gates", + "failed": "Execution failed", + "halted": "Halted instead of guessing", + "idle": "Ready", + "observing": "Observing the application", + "paused": "Execution paused", + "pausing": "Pausing at a safe boundary", + "recording": "Watching your demonstration", + "resuming": "Resuming at a safe boundary", + "rolled_back": "Compensating action completed", + "stopping": "Stopping at a safe boundary", + "verified": "Outcome verified", + "verifying": "Verifying the intended result" + }, + "x-openadapt-terminal-phases": [ + "completed_unverified", + "failed", + "halted", + "rolled_back", + "verified" + ], + "x-openadapt-workflow-label-by-mode": { + "demonstration": "Workflow demonstration", + "governed": "Governed workflow", + "managed": "Managed workflow", + "replay": "Workflow replay" + } +} diff --git a/openadapt_types/schemas/control-overlay-timeline-v1.json b/openadapt_types/schemas/control-overlay-timeline-v1.json new file mode 100644 index 0000000..e599c09 --- /dev/null +++ b/openadapt_types/schemas/control-overlay-timeline-v1.json @@ -0,0 +1,315 @@ +{ + "$defs": { + "ControlOverlayControlsV1": { + "additionalProperties": false, + "properties": { + "pause": { + "default": false, + "title": "Pause", + "type": "boolean" + }, + "resume": { + "default": false, + "title": "Resume", + "type": "boolean" + }, + "stop": { + "default": false, + "title": "Stop", + "type": "boolean" + } + }, + "title": "ControlOverlayControlsV1", + "type": "object" + }, + "ControlOverlayDataClassification": { + "description": "The only classifications allowed to leave a private evidence boundary.", + "enum": [ + "synthetic", + "sanitized_public" + ], + "title": "ControlOverlayDataClassification", + "type": "string" + }, + "ControlOverlayFrameV1": { + "additionalProperties": false, + "description": "A deterministic frame safe for public presentation and composition.", + "properties": { + "controls": { + "$ref": "#/$defs/ControlOverlayControlsV1" + }, + "event_sequence": { + "minimum": 0, + "title": "Event Sequence", + "type": "integer" + }, + "mode": { + "$ref": "#/$defs/ControlOverlayMode" + }, + "observed_at_monotonic_ms": { + "minimum": 0, + "title": "Observed At Monotonic Ms", + "type": "number" + }, + "observed_at_unix_ms": { + "minimum": 0, + "title": "Observed At Unix Ms", + "type": "integer" + }, + "phase": { + "$ref": "#/$defs/ControlOverlayPhase" + }, + "presentation": { + "const": true, + "default": true, + "title": "Presentation", + "type": "boolean" + }, + "profile": { + "anyOf": [ + { + "$ref": "#/$defs/ControlOverlayProfile" + }, + { + "type": "null" + } + ], + "default": null + }, + "schema_version": { + "const": "openadapt.control-overlay-frame/v1", + "default": "openadapt.control-overlay-frame/v1", + "title": "Schema Version", + "type": "string" + }, + "state_id": { + "maxLength": 256, + "minLength": 1, + "title": "State Id", + "type": "string" + }, + "status": { + "maxLength": 64, + "minLength": 1, + "title": "Status", + "type": "string" + }, + "step": { + "$ref": "#/$defs/ControlOverlayStepV1" + }, + "visible": { + "title": "Visible", + "type": "boolean" + }, + "workflow_label": { + "$ref": "#/$defs/ControlOverlayWorkflowLabel" + } + }, + "required": [ + "state_id", + "event_sequence", + "observed_at_unix_ms", + "observed_at_monotonic_ms", + "visible", + "phase", + "workflow_label", + "mode", + "step", + "controls", + "status" + ], + "title": "ControlOverlayFrameV1", + "type": "object", + "x-openadapt-state-id-components": [ + "visibility", + "phase", + "mode", + "profile", + "step.current", + "step.total", + "controls.pause", + "controls.resume", + "controls.stop" + ], + "x-openadapt-status-by-phase": { + "completed_unverified": "Completed without sufficient verification", + "executing": "Executing with verification gates", + "failed": "Execution failed", + "halted": "Halted instead of guessing", + "idle": "Ready", + "observing": "Observing the application", + "paused": "Execution paused", + "pausing": "Pausing at a safe boundary", + "recording": "Watching your demonstration", + "resuming": "Resuming at a safe boundary", + "rolled_back": "Compensating action completed", + "stopping": "Stopping at a safe boundary", + "verified": "Outcome verified", + "verifying": "Verifying the intended result" + }, + "x-openadapt-terminal-phases": [ + "completed_unverified", + "failed", + "halted", + "rolled_back", + "verified" + ], + "x-openadapt-workflow-label-by-mode": { + "demonstration": "Workflow demonstration", + "governed": "Governed workflow", + "managed": "Managed workflow", + "replay": "Workflow replay" + } + }, + "ControlOverlayMode": { + "enum": [ + "demonstration", + "replay", + "governed", + "managed" + ], + "title": "ControlOverlayMode", + "type": "string" + }, + "ControlOverlayPhase": { + "description": "Canonical cross-surface execution and terminal states.", + "enum": [ + "idle", + "observing", + "recording", + "executing", + "pausing", + "paused", + "resuming", + "stopping", + "verifying", + "verified", + "completed_unverified", + "halted", + "failed", + "rolled_back" + ], + "title": "ControlOverlayPhase", + "type": "string" + }, + "ControlOverlayProfile": { + "enum": [ + "demo", + "standard", + "regulated" + ], + "title": "ControlOverlayProfile", + "type": "string" + }, + "ControlOverlayStepV1": { + "additionalProperties": false, + "properties": { + "current": { + "anyOf": [ + { + "minimum": 1, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Current" + }, + "total": { + "anyOf": [ + { + "minimum": 1, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Total" + } + }, + "title": "ControlOverlayStepV1", + "type": "object" + }, + "ControlOverlayTimelineEventV1": { + "additionalProperties": false, + "properties": { + "at_ms": { + "minimum": 0, + "title": "At Ms", + "type": "integer" + }, + "frame": { + "$ref": "#/$defs/ControlOverlayFrameV1" + } + }, + "required": [ + "at_ms", + "frame" + ], + "title": "ControlOverlayTimelineEventV1", + "type": "object" + }, + "ControlOverlayWorkflowLabel": { + "description": "Closed presentation labels; user-authored workflow names are excluded.", + "enum": [ + "Workflow demonstration", + "Workflow replay", + "Governed workflow", + "Managed workflow" + ], + "title": "ControlOverlayWorkflowLabel", + "type": "string" + } + }, + "additionalProperties": false, + "description": "Presentation frames bound to one exact evidence pack and media clip.", + "properties": { + "data_classification": { + "$ref": "#/$defs/ControlOverlayDataClassification" + }, + "duration_ms": { + "exclusiveMinimum": 0, + "title": "Duration Ms", + "type": "integer" + }, + "events": { + "items": { + "$ref": "#/$defs/ControlOverlayTimelineEventV1" + }, + "minItems": 1, + "title": "Events", + "type": "array" + }, + "evidence_pack_id": { + "maxLength": 96, + "minLength": 1, + "pattern": "^[a-z0-9][a-z0-9._-]{0,95}$", + "title": "Evidence Pack Id", + "type": "string" + }, + "media_sha256": { + "pattern": "^[a-f0-9]{64}$", + "title": "Media Sha256", + "type": "string" + }, + "schema_version": { + "const": "openadapt.control-overlay-timeline/v1", + "default": "openadapt.control-overlay-timeline/v1", + "title": "Schema Version", + "type": "string" + } + }, + "required": [ + "data_classification", + "evidence_pack_id", + "media_sha256", + "duration_ms", + "events" + ], + "title": "ControlOverlayTimelineV1", + "type": "object" +} diff --git a/scripts/export_control_overlay_schemas.py b/scripts/export_control_overlay_schemas.py new file mode 100644 index 0000000..1788239 --- /dev/null +++ b/scripts/export_control_overlay_schemas.py @@ -0,0 +1,33 @@ +"""Export the versioned control-overlay JSON Schemas into the package.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from openadapt_types import ControlOverlayFrameV1, ControlOverlayTimelineV1 + +ROOT = Path(__file__).resolve().parents[1] +SCHEMA_DIR = ROOT / "openadapt_types" / "schemas" +SCHEMAS = { + "control-overlay-frame-v1.json": ControlOverlayFrameV1, + "control-overlay-timeline-v1.json": ControlOverlayTimelineV1, +} + + +def rendered_schemas() -> dict[str, str]: + return { + filename: json.dumps(model.model_json_schema(), indent=2, sort_keys=True) + "\n" + for filename, model in SCHEMAS.items() + } + + +def main() -> int: + SCHEMA_DIR.mkdir(parents=True, exist_ok=True) + for filename, text in rendered_schemas().items(): + (SCHEMA_DIR / filename).write_text(text, encoding="utf-8") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_control_overlay.py b/tests/test_control_overlay.py new file mode 100644 index 0000000..7cb165f --- /dev/null +++ b/tests/test_control_overlay.py @@ -0,0 +1,134 @@ +"""Durable contract tests for the cross-surface control overlay.""" + +import json +from importlib.resources import files + +import pytest +from pydantic import ValidationError + +from openadapt_types import ( + CONTROL_OVERLAY_FRAME_SCHEMA, + CONTROL_OVERLAY_TIMELINE_SCHEMA, + ControlOverlayDataClassification, + ControlOverlayFrameV1, + ControlOverlayMode, + ControlOverlayPhase, + ControlOverlayProfile, + ControlOverlayTimelineBindingV1, + ControlOverlayTimelineEventV1, + ControlOverlayTimelineV1, +) + + +def _frame(sequence: int, phase: ControlOverlayPhase) -> ControlOverlayFrameV1: + return ControlOverlayFrameV1.build( + event_sequence=sequence, + observed_at_unix_ms=1_785_000_000_000 + sequence, + observed_at_monotonic_ms=1000.0 + sequence, + visible=True, + phase=phase, + mode=ControlOverlayMode.GOVERNED, + profile=ControlOverlayProfile.REGULATED, + current_step=2, + total_steps=5, + ) + + +def test_frame_builder_derives_canonical_public_state() -> None: + frame = _frame(4, ControlOverlayPhase.EXECUTING) + + assert frame.schema_version == CONTROL_OVERLAY_FRAME_SCHEMA + assert frame.state_id == ( + "visible:executing:governed:regulated:2:5:" "no-pause:no-resume:no-stop" + ) + assert frame.workflow_label == "Governed workflow" + assert frame.status == "Executing with verification gates" + assert frame.presentation is True + + +def test_frame_refuses_free_form_or_inconsistent_presentation_data() -> None: + payload = _frame(1, ControlOverlayPhase.EXECUTING).model_dump(mode="json") + payload["workflow_label"] = "Customer workflow" + with pytest.raises(ValidationError, match="workflow_label"): + ControlOverlayFrameV1.model_validate(payload) + + payload = _frame(1, ControlOverlayPhase.EXECUTING).model_dump(mode="json") + payload["screenshot"] = "secret.png" + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + ControlOverlayFrameV1.model_validate(payload) + + payload = _frame(1, ControlOverlayPhase.EXECUTING).model_dump(mode="json") + payload["state_id"] = "visible:verified" + with pytest.raises(ValidationError, match="state_id"): + ControlOverlayFrameV1.model_validate(payload) + + +def test_timeline_is_ordered_and_bound_to_exact_pack_and_media() -> None: + digest = "a" * 64 + timeline = ControlOverlayTimelineV1( + data_classification=ControlOverlayDataClassification.SYNTHETIC, + evidence_pack_id="mockmed-triage-v2", + media_sha256=digest, + duration_ms=4000, + events=( + ControlOverlayTimelineEventV1( + at_ms=0, + frame=_frame(1, ControlOverlayPhase.EXECUTING), + ), + ControlOverlayTimelineEventV1( + at_ms=3000, + frame=_frame(2, ControlOverlayPhase.VERIFIED), + ), + ), + ) + + assert timeline.schema_version == CONTROL_OVERLAY_TIMELINE_SCHEMA + assert timeline.frame_at(2999).phase == ControlOverlayPhase.EXECUTING + assert timeline.frame_at(3000).phase == ControlOverlayPhase.VERIFIED + timeline.assert_binding( + ControlOverlayTimelineBindingV1( + evidence_pack_id="mockmed-triage-v2", + media_sha256=digest, + ) + ) + with pytest.raises(ValueError, match="exact media digest"): + timeline.assert_binding( + ControlOverlayTimelineBindingV1( + evidence_pack_id="mockmed-triage-v2", + media_sha256="b" * 64, + ) + ) + + payload = timeline.model_dump(mode="json") + payload["events"][1]["at_ms"] = 0 + with pytest.raises(ValidationError, match="offsets must increase"): + ControlOverlayTimelineV1.model_validate(payload) + + +def test_contracts_export_strict_language_agnostic_json_schema() -> None: + frame_schema = ControlOverlayFrameV1.model_json_schema() + timeline_schema = ControlOverlayTimelineV1.model_json_schema() + + assert frame_schema["additionalProperties"] is False + assert timeline_schema["additionalProperties"] is False + assert frame_schema["x-openadapt-status-by-phase"]["verified"] == ( + "Outcome verified" + ) + assert json.dumps(frame_schema).count("openadapt.control-overlay-frame/v1") >= 1 + assert ( + json.dumps(timeline_schema).count("openadapt.control-overlay-timeline/v1") >= 1 + ) + + packaged_schemas = files("openadapt_types.schemas") + assert ( + json.loads( + packaged_schemas.joinpath("control-overlay-frame-v1.json").read_text() + ) + == frame_schema + ) + assert ( + json.loads( + packaged_schemas.joinpath("control-overlay-timeline-v1.json").read_text() + ) + == timeline_schema + )