diff --git a/src/sudachi_life/phase3/__init__.py b/src/sudachi_life/phase3/__init__.py index 904a896..f7b4363 100644 --- a/src/sudachi_life/phase3/__init__.py +++ b/src/sudachi_life/phase3/__init__.py @@ -1,4 +1,4 @@ -"""Deterministic, fixture-only Phase 3 evaluation foundation. +"""Deterministic Phase 3 evaluation and bounded caregiver protocol surfaces. This package is intentionally additive and does not write the Phase 1/2 organism body. """ @@ -22,6 +22,24 @@ validate_proposal, ) from .fixtures import build_valid_fixture_episode +from .human_live import ( + HUMAN_PILOT_LIVE_BRIDGE_VERSION, + LOCAL_STRUCTURED_HUMAN_TRANSPORT, + SELF_HUMAN_ATTEMPT_ID, + SELF_HUMAN_AUTHORIZATION_RECORD, + SELF_HUMAN_CAREGIVER_ID, + SELF_HUMAN_CONSENT_NOTICE_VERSION, + SELF_HUMAN_PILOT_ID, + HumanBridgeResult, + HumanConsultationMeasurement, + HumanPilotAttemptState, + SelfHumanPilotAuthorization, + accept_self_human_proposal, + accepted_self_human_pilot_v1_authorization, + disable_self_human_bridge, + start_self_human_pilot_attempt, + validate_self_human_pilot_authorization, +) from .human_pilot import ( HUMAN_PILOT_PREFLIGHT_VERSION, PROPOSED_MAX_ATTEMPT_WALL_MS, @@ -80,8 +98,10 @@ "ACTIVE_SOURCE_KINDS", "CAREGIVER_PROTOCOL_VERSION", "CONTRACT_VERSION", + "HUMAN_PILOT_LIVE_BRIDGE_VERSION", "HUMAN_PILOT_PREFLIGHT_VERSION", "IMPLEMENTATION_VERSION", + "LOCAL_STRUCTURED_HUMAN_TRANSPORT", "PROPOSED_MAX_ATTEMPT_WALL_MS", "PROPOSED_MAX_CAREGIVER_ACTIVE_MS_PER_ATTEMPT", "PROPOSED_MAX_CLARIFICATIONS_PER_ATTEMPT", @@ -90,6 +110,11 @@ "PROPOSED_MAX_RESPONSE_LATENCY_MS", "PROPOSED_PILOT_ATTEMPTS", "PROPOSED_PROPOSAL_PAYLOAD_RETENTION_DAYS", + "SELF_HUMAN_ATTEMPT_ID", + "SELF_HUMAN_AUTHORIZATION_RECORD", + "SELF_HUMAN_CAREGIVER_ID", + "SELF_HUMAN_CONSENT_NOTICE_VERSION", + "SELF_HUMAN_PILOT_ID", "AccountingStatus", "Availability", "AttemptState", @@ -108,7 +133,10 @@ "FixtureCaregiverAdapter", "FixtureResponse", "HiddenLaborPolicy", + "HumanBridgeResult", "HumanCaregiverPilotPreflight", + "HumanConsultationMeasurement", + "HumanPilotAttemptState", "HumanPilotBudget", "HumanPilotDataPolicy", "HumanPilotPreflightResult", @@ -121,18 +149,24 @@ "ProposalKind", "ProposalValidation", "ReplayConflict", + "SelfHumanPilotAuthorization", + "accept_self_human_proposal", "accepted_phase3_requirement_ids", + "accepted_self_human_pilot_v1_authorization", "advance_attempt_history", "build_integrated_caregiver_rehearsal", "build_valid_fixture_episode", "classify_availability", + "disable_self_human_bridge", "proposal_data_fields", "proposal_from_text", "proposed_human_caregiver_pilot_v1", "reconcile_immutable_replay", + "start_self_human_pilot_attempt", "validate_episode", "validate_human_pilot_preflight", "validate_human_proposal_draft", "validate_integrated_caregiver_rehearsal", "validate_proposal", + "validate_self_human_pilot_authorization", ] diff --git a/src/sudachi_life/phase3/human_live.py b/src/sudachi_life/phase3/human_live.py new file mode 100644 index 0000000..29a3d16 --- /dev/null +++ b/src/sudachi_life/phase3/human_live.py @@ -0,0 +1,449 @@ +from __future__ import annotations + +from dataclasses import dataclass, replace +from datetime import datetime +from threading import RLock + +from .caregiver import ( + AccountingStatus, + CaregiverAccounting, + CaregiverProposal, + CaregiverRequest, + CaregiverSourceKind, + ProposalAuthority, + proposal_from_text, + validate_proposal, +) +from .human_pilot import ( + PROPOSED_MAX_ATTEMPT_WALL_MS, + PROPOSED_MAX_CAREGIVER_ACTIVE_MS_PER_ATTEMPT, + PROPOSED_MAX_CLARIFICATIONS_PER_ATTEMPT, + PROPOSED_MAX_CONSULTATIONS_PER_ATTEMPT, + PROPOSED_MAX_RESPONSE_LATENCY_MS, + PROPOSED_PILOT_ATTEMPTS, + PROPOSED_PROPOSAL_PAYLOAD_RETENTION_DAYS, + EthicsReviewStatus, + HumanProposalDraft, + validate_human_proposal_draft, +) + +HUMAN_PILOT_LIVE_BRIDGE_VERSION = "sudachi.phase3.human_caregiver_live_bridge/v1" +SELF_HUMAN_PILOT_ID = "pilot:human-caregiver-v1-self" +LOCAL_STRUCTURED_HUMAN_TRANSPORT = "local_structured_manual" +SELF_HUMAN_AUTHORIZATION_RECORD = "github:yo4e/sudachi-life/issues/158" +SELF_HUMAN_CONSENT_NOTICE_VERSION = "sudachi.phase3.self_human_consent/v1" +SELF_HUMAN_ATTEMPT_ID = "attempt:self-human-pilot-v1-001" +SELF_HUMAN_CAREGIVER_ID = "caregiver:pseudo:0123456789abcdef0123456789abcdef" + + +@dataclass(frozen=True, slots=True) +class SelfHumanPilotAuthorization: + protocol_version: str + pilot_id: str + attempt_id: str + caregiver_id: str + source_kind: CaregiverSourceKind + transport: str + authorization_record: str + consent_notice_version: str + project_owner_is_caregiver: bool + self_only: bool + ethics_review_status: EthicsReviewStatus + consent_attested: bool + consent_revocable: bool + raw_chat_transcript_retained: bool + public_raw_payload_default: bool + proposal_payload_retention_days: int + third_party_participants: int + + +@dataclass(frozen=True, slots=True) +class HumanPilotAttemptState: + attempt_id: str + generation: int = 0 + consultations_used: int = 0 + clarifications_used: int = 0 + caregiver_active_ms_used: int = 0 + attempt_elapsed_ms: int = 0 + live_enabled: bool = True + disabled_at_elapsed_ms: int | None = None + accepted_request_ids: tuple[str, ...] = () + accepted_proposal_ids: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class HumanConsultationMeasurement: + latency_ms: int + caregiver_active_ms: int + attempt_elapsed_ms: int + source_timestamp: str + is_clarification: bool = False + + +@dataclass(frozen=True, slots=True) +class HumanBridgeResult: + accepted: bool + errors: tuple[str, ...] + proposal: CaregiverProposal | None + state: HumanPilotAttemptState + + +_SESSION_LOCK = RLock() +_AUTHORITATIVE_STATE: HumanPilotAttemptState | None = None + + +def _is_offset_timestamp(value: str) -> bool: + if not value: + return False + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return False + return parsed.tzinfo is not None and parsed.utcoffset() is not None + + +def _validate_attempt_state(state: HumanPilotAttemptState) -> tuple[str, ...]: + errors: list[str] = [] + if state.attempt_id != SELF_HUMAN_ATTEMPT_ID: + errors.append("state.attempt_id") + + integer_fields = ( + ("generation", state.generation, None), + ("consultations_used", state.consultations_used, PROPOSED_MAX_CONSULTATIONS_PER_ATTEMPT), + ("clarifications_used", state.clarifications_used, PROPOSED_MAX_CLARIFICATIONS_PER_ATTEMPT), + ("caregiver_active_ms_used", state.caregiver_active_ms_used, PROPOSED_MAX_CAREGIVER_ACTIVE_MS_PER_ATTEMPT), + ("attempt_elapsed_ms", state.attempt_elapsed_ms, PROPOSED_MAX_ATTEMPT_WALL_MS), + ) + for name, value, maximum in integer_fields: + if type(value) is not int or value < 0 or (maximum is not None and value > maximum): + errors.append(f"state.{name}") + + if type(state.live_enabled) is not bool: + errors.append("state.live_enabled") + if state.clarifications_used > state.consultations_used: + errors.append("state.clarifications_exceed_consultations") + if len(state.accepted_request_ids) != state.consultations_used: + errors.append("state.request_count") + if len(state.accepted_proposal_ids) != state.consultations_used: + errors.append("state.proposal_count") + if len(set(state.accepted_request_ids)) != len(state.accepted_request_ids): + errors.append("state.request_ids_duplicate") + if len(set(state.accepted_proposal_ids)) != len(state.accepted_proposal_ids): + errors.append("state.proposal_ids_duplicate") + if any(not value for value in state.accepted_request_ids): + errors.append("state.request_id_empty") + if any(not value.startswith("proposal:") for value in state.accepted_proposal_ids): + errors.append("state.proposal_id_shape") + + if state.live_enabled: + if state.disabled_at_elapsed_ms is not None: + errors.append("state.enabled_with_disablement") + else: + if type(state.disabled_at_elapsed_ms) is not int: + errors.append("state.disabled_without_timestamp") + elif state.disabled_at_elapsed_ms != state.attempt_elapsed_ms: + errors.append("state.disablement_timestamp") + + return tuple(errors) + + +def accepted_self_human_pilot_v1_authorization() -> SelfHumanPilotAuthorization: + """Return the one exact owner-approved self-only Pilot v1 authorization.""" + return SelfHumanPilotAuthorization( + protocol_version=HUMAN_PILOT_LIVE_BRIDGE_VERSION, + pilot_id=SELF_HUMAN_PILOT_ID, + attempt_id=SELF_HUMAN_ATTEMPT_ID, + caregiver_id=SELF_HUMAN_CAREGIVER_ID, + source_kind=CaregiverSourceKind.HUMAN, + transport=LOCAL_STRUCTURED_HUMAN_TRANSPORT, + authorization_record=SELF_HUMAN_AUTHORIZATION_RECORD, + consent_notice_version=SELF_HUMAN_CONSENT_NOTICE_VERSION, + project_owner_is_caregiver=True, + self_only=True, + ethics_review_status=EthicsReviewStatus.NOT_REQUIRED_DECLARED, + consent_attested=True, + consent_revocable=True, + raw_chat_transcript_retained=False, + public_raw_payload_default=False, + proposal_payload_retention_days=PROPOSED_PROPOSAL_PAYLOAD_RETENTION_DAYS, + third_party_participants=0, + ) + + +def validate_self_human_pilot_authorization( + authorization: SelfHumanPilotAuthorization, +) -> tuple[str, ...]: + errors: list[str] = [] + + if PROPOSED_PILOT_ATTEMPTS != 1: + errors.append("authorization.pilot_attempt_count") + if authorization.protocol_version != HUMAN_PILOT_LIVE_BRIDGE_VERSION: + errors.append("authorization.protocol_version") + if authorization.pilot_id != SELF_HUMAN_PILOT_ID: + errors.append("authorization.pilot_id") + if authorization.attempt_id != SELF_HUMAN_ATTEMPT_ID: + errors.append("authorization.attempt_id") + if authorization.caregiver_id != SELF_HUMAN_CAREGIVER_ID: + errors.append("authorization.caregiver_id") + if authorization.source_kind is not CaregiverSourceKind.HUMAN: + errors.append("authorization.source_kind") + if authorization.transport != LOCAL_STRUCTURED_HUMAN_TRANSPORT: + errors.append("authorization.transport") + if authorization.authorization_record != SELF_HUMAN_AUTHORIZATION_RECORD: + errors.append("authorization.record") + if authorization.consent_notice_version != SELF_HUMAN_CONSENT_NOTICE_VERSION: + errors.append("authorization.consent_notice_version") + if not authorization.project_owner_is_caregiver: + errors.append("authorization.owner_is_caregiver") + if not authorization.self_only: + errors.append("authorization.self_only") + if authorization.ethics_review_status is not EthicsReviewStatus.NOT_REQUIRED_DECLARED: + errors.append("authorization.ethics_review_status") + if not authorization.consent_attested: + errors.append("authorization.consent_attested") + if not authorization.consent_revocable: + errors.append("authorization.consent_revocable") + if authorization.raw_chat_transcript_retained: + errors.append("authorization.raw_chat_transcript") + if authorization.public_raw_payload_default: + errors.append("authorization.public_raw_payload") + if authorization.proposal_payload_retention_days != PROPOSED_PROPOSAL_PAYLOAD_RETENTION_DAYS: + errors.append("authorization.retention_days") + if authorization.third_party_participants != 0: + errors.append("authorization.third_party_participants") + + return tuple(errors) + + +def start_self_human_pilot_attempt( + *, authorization: SelfHumanPilotAuthorization +) -> HumanPilotAttemptState: + """Start the one process-authoritative Pilot v1 attempt exactly once.""" + authorization_errors = validate_self_human_pilot_authorization(authorization) + if authorization_errors: + raise ValueError(f"invalid authorization: {authorization_errors}") + + global _AUTHORITATIVE_STATE + with _SESSION_LOCK: + if _AUTHORITATIVE_STATE is not None: + raise RuntimeError("self-human Pilot v1 attempt already started in this process") + _AUTHORITATIVE_STATE = HumanPilotAttemptState(attempt_id=SELF_HUMAN_ATTEMPT_ID) + return _AUTHORITATIVE_STATE + + +def _current_state_or(state: HumanPilotAttemptState) -> HumanPilotAttemptState: + return _AUTHORITATIVE_STATE if _AUTHORITATIVE_STATE is not None else state + + +def disable_self_human_bridge( + *, + authorization: SelfHumanPilotAuthorization, + state: HumanPilotAttemptState, + attempt_elapsed_ms: int, +) -> HumanPilotAttemptState: + authorization_errors = validate_self_human_pilot_authorization(authorization) + if authorization_errors: + raise ValueError(f"invalid authorization: {authorization_errors}") + + global _AUTHORITATIVE_STATE + with _SESSION_LOCK: + if _AUTHORITATIVE_STATE is None: + raise RuntimeError("self-human Pilot v1 attempt has not started") + if state != _AUTHORITATIVE_STATE: + raise ValueError("stale or forked attempt state") + state_errors = _validate_attempt_state(_AUTHORITATIVE_STATE) + if state_errors: + raise ValueError(f"invalid authoritative attempt state: {state_errors}") + if not _AUTHORITATIVE_STATE.live_enabled: + raise ValueError("bridge is already disabled") + if type(attempt_elapsed_ms) is not int: + raise TypeError("disablement elapsed time must be int") + if attempt_elapsed_ms < _AUTHORITATIVE_STATE.attempt_elapsed_ms: + raise ValueError("disablement elapsed time cannot move backwards") + if attempt_elapsed_ms > PROPOSED_MAX_ATTEMPT_WALL_MS: + raise ValueError("disablement exceeds attempt wall budget") + + _AUTHORITATIVE_STATE = replace( + _AUTHORITATIVE_STATE, + generation=_AUTHORITATIVE_STATE.generation + 1, + attempt_elapsed_ms=attempt_elapsed_ms, + live_enabled=False, + disabled_at_elapsed_ms=attempt_elapsed_ms, + ) + return _AUTHORITATIVE_STATE + + +def accept_self_human_proposal( + *, + authorization: SelfHumanPilotAuthorization, + state: HumanPilotAttemptState, + request: CaregiverRequest, + draft: HumanProposalDraft, + measurement: HumanConsultationMeasurement, + allowed_observation_ids: frozenset[str] = frozenset(), + allowed_objective_ids: frozenset[str] = frozenset(), + allowed_action_ids: frozenset[str] = frozenset(), +) -> HumanBridgeResult: + """Accept one local structured self-caregiver draft against authoritative state. + + The bridge performs no network, subprocess, browser, credential, filesystem, + or external-service access. State progression is process-authoritative and + atomic: stale, fresh-reset, or forked snapshots cannot advance the attempt. + """ + global _AUTHORITATIVE_STATE + with _SESSION_LOCK: + errors = list(validate_self_human_pilot_authorization(authorization)) + if _AUTHORITATIVE_STATE is None: + errors.append("bridge.attempt_not_started") + authoritative = state + else: + authoritative = _AUTHORITATIVE_STATE + if state != authoritative: + errors.append("bridge.stale_or_forked_state") + errors.extend(f"bridge.{error}" for error in _validate_attempt_state(authoritative)) + + if not authoritative.live_enabled: + errors.append("bridge.disabled") + if request.attempt_id != authorization.attempt_id: + errors.append("bridge.attempt_binding") + if request.request_id in authoritative.accepted_request_ids: + errors.append("bridge.request_replay") + if draft.caregiver_id != authorization.caregiver_id: + errors.append("bridge.caregiver_binding") + + draft_validation = validate_human_proposal_draft( + request, + draft, + allowed_observation_ids=allowed_observation_ids, + allowed_objective_ids=allowed_objective_ids, + allowed_action_ids=allowed_action_ids, + ) + errors.extend(f"bridge.{error}" for error in draft_validation.errors) + + measurement_ints = ( + ("latency_ms", measurement.latency_ms), + ("caregiver_active_ms", measurement.caregiver_active_ms), + ("attempt_elapsed_ms", measurement.attempt_elapsed_ms), + ) + for name, value in measurement_ints: + if type(value) is not int: + errors.append(f"bridge.{name}_type") + if type(measurement.is_clarification) is not bool: + errors.append("bridge.is_clarification_type") + + elapsed_delta: int | None = None + if type(measurement.attempt_elapsed_ms) is int: + if measurement.attempt_elapsed_ms < authoritative.attempt_elapsed_ms: + errors.append("bridge.elapsed_time_regression") + elif measurement.attempt_elapsed_ms > PROPOSED_MAX_ATTEMPT_WALL_MS: + errors.append("bridge.attempt_wall_budget") + else: + elapsed_delta = measurement.attempt_elapsed_ms - authoritative.attempt_elapsed_ms + + if type(measurement.latency_ms) is int: + if measurement.latency_ms < 0: + errors.append("bridge.latency_negative") + elif measurement.latency_ms > PROPOSED_MAX_RESPONSE_LATENCY_MS: + errors.append("bridge.latency_budget") + elif elapsed_delta is not None and measurement.latency_ms > elapsed_delta: + errors.append("bridge.latency_exceeds_elapsed") + if type(measurement.caregiver_active_ms) is int: + if measurement.caregiver_active_ms < 0: + errors.append("bridge.active_ms_negative") + elif elapsed_delta is not None and measurement.caregiver_active_ms > elapsed_delta: + errors.append("bridge.active_ms_exceeds_elapsed") + if not _is_offset_timestamp(measurement.source_timestamp): + errors.append("bridge.source_timestamp") + + is_linked_clarification = draft.clarification_of_proposal_id is not None + if type(measurement.is_clarification) is bool and measurement.is_clarification != is_linked_clarification: + errors.append("bridge.clarification_binding") + if is_linked_clarification and draft.clarification_of_proposal_id not in authoritative.accepted_proposal_ids: + errors.append("bridge.clarification_source") + + next_consultations = authoritative.consultations_used + 1 + next_clarifications = authoritative.clarifications_used + int(measurement.is_clarification is True) + next_active_ms = ( + authoritative.caregiver_active_ms_used + measurement.caregiver_active_ms + if type(measurement.caregiver_active_ms) is int + else authoritative.caregiver_active_ms_used + ) + + if next_consultations > PROPOSED_MAX_CONSULTATIONS_PER_ATTEMPT: + errors.append("bridge.consultation_budget") + if next_clarifications > PROPOSED_MAX_CLARIFICATIONS_PER_ATTEMPT: + errors.append("bridge.clarification_budget") + if next_active_ms > PROPOSED_MAX_CAREGIVER_ACTIVE_MS_PER_ATTEMPT: + errors.append("bridge.active_time_budget") + + if errors: + return HumanBridgeResult( + accepted=False, + errors=tuple(errors), + proposal=None, + state=_current_state_or(state), + ) + + accounting = CaregiverAccounting( + consultation_count=1, + clarification_count=int(measurement.is_clarification), + latency_ms=measurement.latency_ms, + human_active_ms=measurement.caregiver_active_ms, + model_calls=None, + money_minor_units=None, + live_cost_status=AccountingStatus.MEASURED, + absence_reason=None, + ) + proposal = proposal_from_text( + request=request, + source_id=authorization.caregiver_id, + source_kind=CaregiverSourceKind.HUMAN, + kind=draft.kind, + text=draft.text, + accounting=accounting, + source_timestamp=measurement.source_timestamp, + ) + + generic_validation = validate_proposal(request, proposal) + unexpected_generic_errors = tuple( + error + for error in generic_validation.errors + if error != "proposal.live_source_not_authorized" + ) + if "proposal.live_source_not_authorized" not in generic_validation.errors: + unexpected_generic_errors += ("proposal.expected_global_rejection_missing",) + if proposal.proposal_id in authoritative.accepted_proposal_ids: + unexpected_generic_errors += ("proposal.replay",) + if unexpected_generic_errors: + return HumanBridgeResult( + accepted=False, + errors=tuple(f"bridge.{error}" for error in unexpected_generic_errors), + proposal=None, + state=authoritative, + ) + + if proposal.authority is not ProposalAuthority.PROPOSAL_ONLY: + return HumanBridgeResult(False, ("bridge.proposal_authority",), None, authoritative) + if proposal.source_kind is not CaregiverSourceKind.HUMAN: + return HumanBridgeResult(False, ("bridge.proposal_source_kind",), None, authoritative) + if draft_validation.payload_sha256 != proposal.payload_sha256: + return HumanBridgeResult(False, ("bridge.payload_digest",), None, authoritative) + + _AUTHORITATIVE_STATE = HumanPilotAttemptState( + attempt_id=SELF_HUMAN_ATTEMPT_ID, + generation=authoritative.generation + 1, + consultations_used=next_consultations, + clarifications_used=next_clarifications, + caregiver_active_ms_used=next_active_ms, + attempt_elapsed_ms=measurement.attempt_elapsed_ms, + live_enabled=True, + disabled_at_elapsed_ms=None, + accepted_request_ids=authoritative.accepted_request_ids + (request.request_id,), + accepted_proposal_ids=authoritative.accepted_proposal_ids + (proposal.proposal_id,), + ) + return HumanBridgeResult( + accepted=True, + errors=(), + proposal=proposal, + state=_AUTHORITATIVE_STATE, + ) diff --git a/tests/test_phase3_self_human_live_bridge.py b/tests/test_phase3_self_human_live_bridge.py new file mode 100644 index 0000000..eaabf40 --- /dev/null +++ b/tests/test_phase3_self_human_live_bridge.py @@ -0,0 +1,504 @@ +from __future__ import annotations + +from dataclasses import replace + +import pytest + +import sudachi_life.phase3.human_live as live_module +from sudachi_life.phase3.caregiver import ( + ACTIVE_SOURCE_KINDS, + AccountingStatus, + CaregiverRequest, + CaregiverSourceKind, + ProposalAuthority, + ProposalKind, + validate_proposal, +) +from sudachi_life.phase3.human_live import ( + HUMAN_PILOT_LIVE_BRIDGE_VERSION, + LOCAL_STRUCTURED_HUMAN_TRANSPORT, + SELF_HUMAN_ATTEMPT_ID, + SELF_HUMAN_AUTHORIZATION_RECORD, + SELF_HUMAN_CAREGIVER_ID, + SELF_HUMAN_CONSENT_NOTICE_VERSION, + SELF_HUMAN_PILOT_ID, + HumanConsultationMeasurement, + HumanPilotAttemptState, + accept_self_human_proposal, + accepted_self_human_pilot_v1_authorization, + disable_self_human_bridge, + start_self_human_pilot_attempt, + validate_self_human_pilot_authorization, +) +from sudachi_life.phase3.human_pilot import ( + PROPOSED_MAX_ATTEMPT_WALL_MS, + PROPOSED_MAX_CAREGIVER_ACTIVE_MS_PER_ATTEMPT, + PROPOSED_MAX_CLARIFICATIONS_PER_ATTEMPT, + PROPOSED_MAX_CONSULTATIONS_PER_ATTEMPT, + PROPOSED_MAX_RESPONSE_LATENCY_MS, + CaregiverConfidence, + EthicsReviewStatus, + HumanProposalDraft, +) + + +@pytest.fixture(autouse=True) +def _reset_process_authoritative_state(): + with live_module._SESSION_LOCK: + live_module._AUTHORITATIVE_STATE = None + yield + with live_module._SESSION_LOCK: + live_module._AUTHORITATIVE_STATE = None + + +def _authorization(): + return accepted_self_human_pilot_v1_authorization() + + +def _request(*, sequence_ordinal: int = 1) -> CaregiverRequest: + return CaregiverRequest( + request_id=f"request:self-human-{sequence_ordinal:03d}", + study_id="study:self-human-pilot-v1", + attempt_id=SELF_HUMAN_ATTEMPT_ID, + episode_id="episode:self-human-pilot-v1-001", + organism_id="sudachi-self-human-v1", + lineage_generation=0, + sequence_ordinal=sequence_ordinal, + allowed_kinds=(ProposalKind.EXPLANATION, ProposalKind.ABSTAIN), + ) + + +def _draft(*, sequence_ordinal: int = 1, **changes: object) -> HumanProposalDraft: + values: dict[str, object] = { + "draft_id": f"draft:self-human-{sequence_ordinal:03d}", + "request_id": f"request:self-human-{sequence_ordinal:03d}", + "caregiver_id": SELF_HUMAN_CAREGIVER_ID, + "sequence_ordinal": sequence_ordinal, + "kind": ProposalKind.EXPLANATION, + "text": "The visible marker supports inspecting before moving.", + "confidence": CaregiverConfidence.MEDIUM, + "observation_ids": ("observation:marker-visible",), + "objective_ids": ("objective:reach-marker",), + "action_ids": ("action:inspect", "action:move"), + } + values.update(changes) + return HumanProposalDraft(**values) # type: ignore[arg-type] + + +def _measurement( + *, + elapsed_ms: int = 30_000, + latency_ms: int = 2_000, + active_ms: int = 5_000, + is_clarification: bool = False, + timestamp: str = "2026-08-30T14:30:00+09:00", +) -> HumanConsultationMeasurement: + return HumanConsultationMeasurement( + latency_ms=latency_ms, + caregiver_active_ms=active_ms, + attempt_elapsed_ms=elapsed_ms, + source_timestamp=timestamp, + is_clarification=is_clarification, + ) + + +def _start(): + return start_self_human_pilot_attempt(authorization=_authorization()) + + +def _accept( + *, + authorization=None, + state: HumanPilotAttemptState | None = None, + request: CaregiverRequest | None = None, + draft: HumanProposalDraft | None = None, + measurement: HumanConsultationMeasurement | None = None, +): + actual_state = state if state is not None else _start() + return accept_self_human_proposal( + authorization=authorization or _authorization(), + state=actual_state, + request=request or _request(), + draft=draft or _draft(), + measurement=measurement or _measurement(), + allowed_observation_ids=frozenset({"observation:marker-visible"}), + allowed_objective_ids=frozenset({"objective:reach-marker"}), + allowed_action_ids=frozenset({"action:inspect", "action:move"}), + ) + + +def test_authorization_is_one_exact_attempt_and_caregiver() -> None: + authorization = _authorization() + + assert authorization.protocol_version == HUMAN_PILOT_LIVE_BRIDGE_VERSION + assert authorization.pilot_id == SELF_HUMAN_PILOT_ID + assert authorization.attempt_id == SELF_HUMAN_ATTEMPT_ID + assert authorization.caregiver_id == SELF_HUMAN_CAREGIVER_ID + assert authorization.transport == LOCAL_STRUCTURED_HUMAN_TRANSPORT + assert authorization.authorization_record == SELF_HUMAN_AUTHORIZATION_RECORD + assert authorization.consent_notice_version == SELF_HUMAN_CONSENT_NOTICE_VERSION + assert authorization.project_owner_is_caregiver is True + assert authorization.self_only is True + assert authorization.ethics_review_status is EthicsReviewStatus.NOT_REQUIRED_DECLARED + assert authorization.consent_attested is True + assert authorization.consent_revocable is True + assert authorization.third_party_participants == 0 + assert validate_self_human_pilot_authorization(authorization) == () + + +def test_second_attempt_or_caregiver_cannot_mint_equivalent_authorization() -> None: + authorization = _authorization() + + wrong_attempt = replace(authorization, attempt_id="attempt:self-human-pilot-v1-002") + wrong_caregiver = replace( + authorization, + caregiver_id="caregiver:pseudo:ffffffffffffffffffffffffffffffff", + ) + + assert "authorization.attempt_id" in validate_self_human_pilot_authorization(wrong_attempt) + assert "authorization.caregiver_id" in validate_self_human_pilot_authorization(wrong_caregiver) + + +def test_attempt_start_is_process_authoritative_and_one_shot() -> None: + authorization = _authorization() + first = start_self_human_pilot_attempt(authorization=authorization) + + assert first.attempt_id == SELF_HUMAN_ATTEMPT_ID + assert first.generation == 0 + with pytest.raises(RuntimeError): + start_self_human_pilot_attempt(authorization=authorization) + + +def test_live_bridge_accepts_structured_self_human_proposal_only() -> None: + state = _start() + result = _accept(state=state) + + assert result.accepted is True + assert result.errors == () + assert result.proposal is not None + proposal = result.proposal + assert proposal.source_kind is CaregiverSourceKind.HUMAN + assert proposal.source_id == SELF_HUMAN_CAREGIVER_ID + assert proposal.authority is ProposalAuthority.PROPOSAL_ONLY + assert proposal.accounting.live_cost_status is AccountingStatus.MEASURED + assert proposal.accounting.human_active_ms == 5_000 + assert proposal.accounting.latency_ms == 2_000 + assert proposal.accounting.model_calls is None + assert proposal.accounting.money_minor_units is None + assert result.state.generation == 1 + assert result.state.consultations_used == 1 + assert result.state.accepted_request_ids == (_request().request_id,) + assert result.state.accepted_proposal_ids == (proposal.proposal_id,) + + +def test_generic_validator_remains_fixture_only() -> None: + state = _start() + result = _accept(state=state) + assert result.proposal is not None + + generic = validate_proposal(_request(), result.proposal) + + assert generic.valid is False + assert generic.errors == ("proposal.live_source_not_authorized",) + assert ACTIVE_SOURCE_KINDS == frozenset({CaregiverSourceKind.FIXTURE}) + assert CaregiverSourceKind.HUMAN not in ACTIVE_SOURCE_KINDS + + +def test_fresh_state_reset_is_rejected_after_progress() -> None: + state0 = _start() + first = _accept(state=state0) + assert first.accepted is True + + forged_fresh = HumanPilotAttemptState(attempt_id=SELF_HUMAN_ATTEMPT_ID) + replay = _accept( + state=forged_fresh, + request=_request(sequence_ordinal=2), + draft=_draft(sequence_ordinal=2), + measurement=_measurement(elapsed_ms=40_000), + ) + + assert replay.accepted is False + assert "bridge.stale_or_forked_state" in replay.errors + assert replay.state == first.state + + +def test_parallel_fork_from_same_prior_state_is_rejected() -> None: + state0 = _start() + first = _accept(state=state0) + assert first.accepted is True + + fork = _accept( + state=state0, + request=_request(sequence_ordinal=2), + draft=_draft(sequence_ordinal=2), + measurement=_measurement(elapsed_ms=40_000), + ) + + assert fork.accepted is False + assert "bridge.stale_or_forked_state" in fork.errors + assert fork.state == first.state + + +def test_stale_enabled_state_cannot_bypass_disablement() -> None: + state0 = _start() + first = _accept(state=state0) + disabled = disable_self_human_bridge( + authorization=_authorization(), + state=first.state, + attempt_elapsed_ms=40_000, + ) + + stale_after_disable = _accept( + state=first.state, + request=_request(sequence_ordinal=2), + draft=_draft(sequence_ordinal=2), + measurement=_measurement(elapsed_ms=50_000), + ) + + assert disabled.live_enabled is False + assert stale_after_disable.accepted is False + assert "bridge.stale_or_forked_state" in stale_after_disable.errors + assert "bridge.disabled" in stale_after_disable.errors + assert stale_after_disable.state == disabled + + +def test_current_disabled_state_rejects_post_cutoff_submission() -> None: + state0 = _start() + disabled = disable_self_human_bridge( + authorization=_authorization(), + state=state0, + attempt_elapsed_ms=10_000, + ) + result = _accept( + state=disabled, + measurement=_measurement(elapsed_ms=20_000), + ) + + assert result.accepted is False + assert "bridge.disabled" in result.errors + assert result.state == disabled + + +def test_clarification_parent_must_be_actually_accepted() -> None: + state0 = _start() + first = _accept(state=state0) + assert first.proposal is not None + + unknown = _draft( + sequence_ordinal=2, + clarification_of_proposal_id="proposal:" + "f" * 64, + ) + rejected = _accept( + state=first.state, + request=_request(sequence_ordinal=2), + draft=unknown, + measurement=_measurement(elapsed_ms=40_000, is_clarification=True), + ) + assert "bridge.clarification_source" in rejected.errors + + linked = _draft( + sequence_ordinal=2, + clarification_of_proposal_id=first.proposal.proposal_id, + ) + accepted = _accept( + state=first.state, + request=_request(sequence_ordinal=2), + draft=linked, + measurement=_measurement(elapsed_ms=40_000, is_clarification=True), + ) + assert accepted.accepted is True + assert accepted.state.clarifications_used == 1 + + +def test_forged_well_shaped_history_is_not_authoritative() -> None: + state0 = _start() + first = _accept(state=state0) + forged = HumanPilotAttemptState( + attempt_id=SELF_HUMAN_ATTEMPT_ID, + generation=1, + consultations_used=1, + caregiver_active_ms_used=5_000, + attempt_elapsed_ms=30_000, + accepted_request_ids=("request:forged",), + accepted_proposal_ids=("proposal:" + "a" * 64,), + ) + + result = _accept( + state=forged, + request=_request(sequence_ordinal=2), + draft=_draft( + sequence_ordinal=2, + clarification_of_proposal_id="proposal:" + "a" * 64, + ), + measurement=_measurement(elapsed_ms=40_000, is_clarification=True), + ) + + assert result.accepted is False + assert "bridge.stale_or_forked_state" in result.errors + assert result.state == first.state + + +def test_rejected_submission_does_not_advance_authoritative_state() -> None: + state0 = _start() + rejected = _accept(state=state0, draft=_draft(contains_secret=True)) + + assert rejected.accepted is False + assert rejected.state == state0 + + valid = _accept(state=state0) + assert valid.accepted is True + assert valid.state.generation == 1 + + +def test_request_replay_is_rejected_on_current_state() -> None: + state0 = _start() + first = _accept(state=state0) + replay = _accept( + state=first.state, + measurement=_measurement(elapsed_ms=40_000), + ) + + assert replay.accepted is False + assert "bridge.request_replay" in replay.errors + + +def test_budget_and_elapsed_time_controls_fail_closed() -> None: + state0 = _start() + + latency = _accept( + state=state0, + measurement=_measurement( + elapsed_ms=PROPOSED_MAX_RESPONSE_LATENCY_MS + 2, + latency_ms=PROPOSED_MAX_RESPONSE_LATENCY_MS + 1, + ), + ) + active = _accept( + state=state0, + measurement=_measurement( + elapsed_ms=PROPOSED_MAX_CAREGIVER_ACTIVE_MS_PER_ATTEMPT + 2, + active_ms=PROPOSED_MAX_CAREGIVER_ACTIVE_MS_PER_ATTEMPT + 1, + ), + ) + wall = _accept( + state=state0, + measurement=_measurement(elapsed_ms=PROPOSED_MAX_ATTEMPT_WALL_MS + 1), + ) + + assert "bridge.latency_budget" in latency.errors + assert "bridge.active_time_budget" in active.errors + assert "bridge.attempt_wall_budget" in wall.errors + + +def test_measured_time_must_fit_elapsed_delta() -> None: + state0 = _start() + latency = _accept( + state=state0, + measurement=_measurement(elapsed_ms=1_000, latency_ms=1_001, active_ms=500), + ) + active = _accept( + state=state0, + measurement=_measurement(elapsed_ms=1_000, latency_ms=500, active_ms=1_001), + ) + + assert "bridge.latency_exceeds_elapsed" in latency.errors + assert "bridge.active_ms_exceeds_elapsed" in active.errors + + +def test_privacy_visible_reference_and_caregiver_identity_violations_reject() -> None: + state0 = _start() + draft = _draft( + caregiver_id="caregiver:pseudo:ffffffffffffffffffffffffffffffff", + contains_personal_data=True, + contains_secret=True, + contains_third_party_confidential_data=True, + observation_ids=("observation:hidden",), + ) + result = _accept(state=state0, draft=draft) + + assert "bridge.caregiver_binding" in result.errors + assert "bridge.draft.personal_data_forbidden" in result.errors + assert "bridge.draft.secret_forbidden" in result.errors + assert "bridge.draft.third_party_confidential_data_forbidden" in result.errors + assert "bridge.draft.observation_ids.not_allowed" in result.errors + + +def test_wrong_attempt_and_relaxed_authorization_reject() -> None: + state0 = _start() + wrong_request = replace(_request(), attempt_id="attempt:other") + wrong_auth = replace( + _authorization(), + self_only=False, + project_owner_is_caregiver=False, + ethics_review_status=EthicsReviewStatus.UNRESOLVED, + third_party_participants=1, + ) + + request_result = _accept(state=state0, request=wrong_request) + authorization_result = _accept(state=state0, authorization=wrong_auth) + + assert "bridge.attempt_binding" in request_result.errors + assert "authorization.self_only" in authorization_result.errors + assert "authorization.owner_is_caregiver" in authorization_result.errors + assert "authorization.ethics_review_status" in authorization_result.errors + assert "authorization.third_party_participants" in authorization_result.errors + + +def test_timestamp_and_integer_types_fail_closed() -> None: + state0 = _start() + naive = _accept( + state=state0, + measurement=_measurement(timestamp="2026-08-30T14:30:00"), + ) + bool_latency = _accept( + state=state0, + measurement=HumanConsultationMeasurement( + latency_ms=True, # type: ignore[arg-type] + caregiver_active_ms=1, + attempt_elapsed_ms=10, + source_timestamp="2026-08-30T14:30:00+09:00", + ), + ) + + assert "bridge.source_timestamp" in naive.errors + assert "bridge.latency_ms_type" in bool_latency.errors + + +def test_disablement_is_one_way_and_time_bounded() -> None: + state0 = _start() + state1 = _accept(state=state0).state + + with pytest.raises(ValueError): + disable_self_human_bridge( + authorization=_authorization(), + state=state0, + attempt_elapsed_ms=40_000, + ) + with pytest.raises(ValueError): + disable_self_human_bridge( + authorization=_authorization(), + state=state1, + attempt_elapsed_ms=state1.attempt_elapsed_ms - 1, + ) + + disabled = disable_self_human_bridge( + authorization=_authorization(), + state=state1, + attempt_elapsed_ms=40_000, + ) + with pytest.raises(ValueError): + disable_self_human_bridge( + authorization=_authorization(), + state=disabled, + attempt_elapsed_ms=50_000, + ) + + +def test_public_surface_contains_no_direct_execution_authority() -> None: + authorization = _authorization() + state = _start() + + for obj in (authorization, state): + assert not hasattr(obj, "action") + assert not hasattr(obj, "writer") + assert not hasattr(obj, "execute") + assert not hasattr(obj, "command")