informed-decision/tests/test_lifecycle_review.py

import base64
from dataclasses import replace
import json
import time

import pytest

from informed_decision.disposition import Actor, ActorKind, Verb, DispositionRefused
from informed_decision.lifecycle import Attribution, LifecycleRepository
from informed_decision.lifecycle_service import Caller, LifecycleService, LifecycleError
from informed_decision.policy import build_request
from informed_decision.provenance import Claim, Route
from informed_decision.records import memo_from
from informed_decision.store import Conflict, Store
from test_intake import envelope
from test_intake_service import setup, request
from test_policy_client import policy
from test_browser_auth import signing_key


@pytest.fixture
def handoff(setup):
    intake, store, calls, configuration = setup
    env = envelope()
    memo = memo_from(env['memo'])
    app = LifecycleService(LifecycleRepository(store), policy())
    actor = Caller(Attribution('tenant:platform', Actor('author-agent', ActorKind.AGENT)), time.time()+3600, '{}')
    documents = [{'digest': memo.packet[0].hash, 'media_type': 'text/markdown',
                  'content_base64': base64.b64encode(env['packet'][0]['content'].encode()).decode()}]
    app.register(actor, memo, key='register', documents=documents)
    app.publish(actor, memo.id, 1, audience={'subjects': ['keycape-human'], 'groups': []}, key='publish')
    options = {'submitter': 'state-hub', 'memo_class': env['memo_class'], 'review_group': env['review_group'],
               'approval_intent': env['approval_intent']}
    return app, actor, memo, options, intake, calls


def prepare(handoff):
    app, actor, memo, options, intake, calls = handoff
    return app.request_review(actor, memo.id, 1, options=options, key='request-review')


def present(store, review_id):
    return store.present(review_id, principal_sub='keycape-human',
                         tenant=Claim('tenant:platform', Route.REGISTRATION),
                         principal_type=Claim('human', Route.AUTHENTICATION))


def test_durable_handoff_reuses_intake_and_exposes_revision_outcomes(handoff):
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    app.repository.store = Store(app.repository.store.path)
    assert prepare(handoff) == receipt
    assert app.outcomes(actor, memo.id, 1)['review']['intake_state'] == 'not_submitted'
    assert request(intake, receipt['envelope'])[0] == 201
    assert request(intake, receipt['envelope'])[0] == 200
    assert len(calls) == 1
    store = app.repository.store
    review = store.memo(receipt['review_memo_id'])
    assert review.brief == memo.brief and review.packet == memo.packet
    assert store.memo(memo.id).approval_id is None
    p = present(store, review.id)
    human = Actor('keycape-human', ActorKind.PERSON)
    store.acknowledge(p.id, human, review.required_ack_ids)
    d = store.record_disposition(p.id, human, Verb.DECLINE, operation_id='answer', reasons=('wrong_scope',))
    result = app.outcomes(actor, memo.id, 1)
    assert result['review']['path'] == '/review?memo_id=' + review.id
    assert result['dispositions'][0]['disposition']['id'] == d.id
    assert result['dispositions'][0]['submission_state'] is None
    app.revise(actor, memo.next_version(brief='Refined'), expected_version=1, summary='Update', key='revise')
    assert app.outcomes(actor, memo.id, 2) == {'review': None, 'dispositions': []}
    assert app.outcomes(actor, memo.id, 1)['dispositions'] == result['dispositions']


def test_altered_handoff_and_wrong_submitter_fail_before_create(handoff):
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    changed = json.loads(json.dumps(receipt['envelope']))
    changed['memo']['brief'] += ' Altered'
    assert request(intake, changed) == (409, {'error': 'review_handoff_conflict'})
    data = json.loads(intake.submitters_path.read_text())
    for row in data.values():
        row['submitter_id'] = 'another'
    intake.submitters_path.write_text(json.dumps(data))
    assert request(intake, receipt['envelope'])[0] == 409
    assert not calls


def test_revision_invalidates_presented_review_and_prepared_accept(handoff):
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    assert request(intake, receipt['envelope'])[0] == 201
    store = app.repository.store
    p = present(store, receipt['review_memo_id'])
    human = Actor('keycape-human', ActorKind.PERSON)
    store.acknowledge(p.id, human, memo.required_ack_ids)
    d = store.record_disposition(p.id, human, Verb.ACCEPT, operation_id='accept')
    app.revise(actor, memo.next_version(brief='Revised'), expected_version=1, summary='Update', key='revise')
    with pytest.raises(Conflict):
        store.begin_submission(d.id)
    with pytest.raises(Conflict):
        store.record_disposition(p.id, human, Verb.DECLINE, operation_id='late-answer', reasons=('wrong_scope',))
    with pytest.raises(Conflict):
        present(store, receipt['review_memo_id'])
    assert request(intake, receipt['envelope'])[0] == 409


def test_unresolved_entry_freezes_source_and_snapshot_import_is_refused(handoff):
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    assert request(intake, receipt['envelope'])[0] == 201
    store = app.repository.store
    review = store.memo(receipt['review_memo_id'])
    with pytest.raises(Conflict):
        store.save_memo(review.next_version(brief='Bypass'))
    with pytest.raises(Conflict):
        store.save_memo(review)
    p = present(store, review.id)
    human = Actor('keycape-human', ActorKind.PERSON)
    store.acknowledge(p.id, human, memo.required_ack_ids)
    d = store.record_disposition(p.id, human, Verb.ACCEPT, operation_id='accept')
    attempt = store.begin_submission(d.id)
    for unresolved in (False, True):
        if unresolved:
            store.finish_submission(d.id, attempt)
        with pytest.raises(Conflict):
            app.revise(actor, memo.next_version(brief='Update'), expected_version=1, summary='Update', key='revise')
    assert store.memo(memo.id).version == 1


def test_denial_replay_conflict_and_agent_binding_guard(handoff):
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    with pytest.raises(Conflict):
        app.request_review(actor, memo.id, 1, options={**options, 'review_group': 'other'}, key='request-review')
    with pytest.raises(Conflict):
        app.request_review(actor, memo.id, 1, options=options, key='duplicate')
    app.policy.transport.change = lambda data: data.update(effect='deny')
    with pytest.raises(LifecycleError):
        prepare(handoff)
    with pytest.raises(LifecycleError):
        app.outcomes(actor, memo.id, 1)
    assert request(intake, receipt['envelope'])[0] == 201
    p = present(app.repository.store, receipt['review_memo_id'])
    with pytest.raises(DispositionRefused):
        app.repository.store.record_disposition(p.id, Actor('keycape-human', ActorKind.AGENT), Verb.ACCEPT,
                                                 operation_id='agent-accept')


def test_create_reply_loss_does_not_retry_and_source_change_before_load_refuses(handoff):
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    creator = intake.creator
    def raced(body):
        result = creator(body)
        app.revise(actor, memo.next_version(brief='Changed during create'), expected_version=1, summary='Update', key='revise')
        return result
    intake.creator = raced
    assert request(intake, receipt['envelope'])[0] == 409
    assert request(intake, receipt['envelope'])[0] == 409
    assert len(calls) == 1
    assert app.outcomes(actor, memo.id, 1)['review']['intake_state'] == 'created'
    with app.repository.store._connection() as db:
        assert not db.execute('SELECT 1 FROM memos WHERE id=?', (receipt['review_memo_id'],)).fetchone()

from test_approval_component import component, signing_key
from test_durable_component import receiver


@pytest.mark.parametrize("via_dispatch", [False, True])
def test_actual_engine_human_entry_and_independent_audit_link_to_source_revision(handoff, component, receiver, tmp_path, via_dispatch):
    import jwt
    from informed_decision.intake_approval import IntakeApprovalClient
    from informed_decision.approval_http import ApprovalHTTPClient
    from informed_decision.review import ReviewController
    from informed_decision.runtime import AuditPump
    app, actor, memo, options, intake, calls = handoff
    if via_dispatch:
        import os, sys
        source = os.environ.get('INFD_STATE_HUB_SOURCE')
        if not source:
            pytest.skip('State Hub source required for private dispatcher integration')
        sys.path.insert(0, source)
        from api.services.memo_dispatch import Dispatcher
        options = {**options, 'memo_class': 'decision-record'}
        grants = json.loads(intake.submitters_path.read_text())
        for row in grants.values():
            row['memo_classes'] = ['decision-record']
        intake.submitters_path.write_text(json.dumps(grants))
    client, engine, transport, human, signing = component
    groups = json.loads(intake.groups_path.read_text())
    groups[options['review_group']]['subject'] = human.subject
    intake.groups_path.write_text(json.dumps(groups))
    claims = jwt.decode(human.access_token, options={'verify_signature': False})
    claims.update(sub='informed-decision', principal_type='service', scope='approval:create')
    token = jwt.encode(claims, signing, algorithm='RS256', headers={'kid': 'key-1'})
    intake.creator = IntakeApprovalClient('https://approval.test', lambda: token, transport=transport)
    app.revise(actor, memo.next_version(brief='Refined for human acceptance'), expected_version=1,
               summary='Responded to research', key='revise')
    app.publish(actor, memo.id, 2, audience={'subjects': [human.subject], 'groups': []}, key='publish-second')
    receipt = app.request_review(actor, memo.id, 2, options=options, key='review-second')
    if via_dispatch:
        attempts = []
        now = [time.time()]
        def send(method, url, body, headers):
            assert headers['Authorization'] == 'Bearer projected-token'
            result = request(intake, body)
            attempts.append(result[0])
            if len(attempts) == 1:
                raise OSError('lost reply after native creation')
            return result
        worker = Dispatcher(tmp_path / 'private-dispatch', send=send,
            read_token=lambda: 'projected-token', clock=lambda: now[0])
        assert worker.enqueue(receipt['envelope']).status == 'pending'
        assert not attempts
        assert worker.once().status == 'pending'
        now[0] += 10
        worker = Dispatcher(tmp_path / 'private-dispatch', send=send,
            read_token=lambda: 'projected-token', clock=lambda: now[0])
        forwarded = worker.once()
        assert forwarded.status == 'delivered' and attempts == [201, 200]
        loaded = forwarded.receipt
        assert worker.once() is None
    else:
        status, loaded = request(intake, receipt['envelope'])
        assert status == 201, loaded
    controller = ReviewController(app.repository.store, policy(),
        lambda session: ApprovalHTTPClient('https://approval.test', session, transport=transport), intake_attributes=True)
    page = controller.open(human, receipt['review_memo_id'])
    context = controller.policy.transport.calls[-1][3]['context']
    assert context['intake_submitter'] == 'state-hub'
    assert context['approval_requester'] == 'informed-decision' and context['approval_human_control'] is True
    controller.acknowledge(human, page.presentation.id, memo.required_ack_ids)
    controller.act(human, page.presentation.id, Verb.ACCEPT, operation_id='human-accept')
    result = app.outcomes(actor, memo.id, 2)
    assert result['dispositions'][0]['submission_state'] == 'confirmed'
    from informed_decision.web import App, SESSION_COOKIE
    from test_browser_auth import call
    class Login:
        def session(self, sid):
            return human if sid == 'fixture' else None
    browser = App(Login(), controller, lifecycle=app)
    page_response = call(browser, '/decisions/' + memo.id, query='version=2', cookie=SESSION_COOKIE + '=fixture')
    assert page_response['status'] == 200
    assert 'Open protected human review' in page_response['body']
    assert 'approval entry confirmed' in page_response['body']
    native_page = call(browser, '/presentations/' + page.presentation.id, cookie=SESSION_COOKIE + '=fixture')
    assert native_page['status'] == 200 and 'decision revision 2' in native_page['body']
    assert 'content_base64' not in page_response['body']
    assert len(engine.get(loaded['approval_id']).entries) == 1
    assert app.outcomes(actor, memo.id, 1)['dispositions'] == []
    source = app.repository.store.presentation_content(page.presentation.id)['lifecycle_source']
    assert source == {'memo_id': memo.id, 'version': 2, 'request_event_id': receipt['event_id']}
    # Both lifecycle commitments and the existing presentation/disposition evidence drain.
    pump = AuditPump(app.repository.store, receiver[0], lifecycle_delivery=True)
    for _ in range(3):
        pump.tick()
    assert app.repository.store.lifecycle_counts()['pending'] == 0
    assert all(row['state'] == 'delivered' for row in app.repository.store.outbox())
    assert not any(url.endswith('/consume') for _, url in transport.calls)


def test_http_review_request_and_outcome_adapter(handoff):
    from informed_decision.lifecycle_http import LifecycleAPI
    from test_lifecycle_http import call
    app, actor, memo, options, intake, calls = handoff
    api = LifecycleAPI(app, lambda token: actor)
    status, receipt, _ = call(api, f'/v1/decisions/{memo.id}/review-requests', method='POST',
                               body={'version': 1, 'options': options})
    assert status == 200
    assert request(intake, receipt['envelope'])[0] == 201
    status, outcome, _ = call(api, f'/v1/decisions/{memo.id}/outcomes', query='version=1')
    assert status == 200 and outcome['review']['intake_state'] == 'ready'
    assert call(api, f'/v1/decisions/{memo.id}/outcomes')[0] == 400


def test_review_request_requires_publication_and_supported_profile(handoff):
    app, actor, memo, options, intake, calls = handoff
    app.revise(actor, memo.next_version(brief='Unpublished successor'), expected_version=1, summary='Update', key='revise')
    with pytest.raises(Conflict, match='publish'):
        app.request_review(actor, memo.id, 2, options=options, key='unpublished')
    from informed_decision.memo import StepKind
    app.revise(actor, app.retrieve(actor, memo.id).next_version(step_kind=StepKind.MITZEICHNUNG),
               expected_version=2, summary='Other step', key='revise-step')
    app.publish(actor, memo.id, 3, audience={'subjects': ['keycape-human'], 'groups': []}, key='publish-third')
    with pytest.raises(ValueError, match='profile'):
        app.request_review(actor, memo.id, 3, options=options, key='unsupported')
    assert not calls
    assert not any(event['kind'] == 'review_requested' for event in app.events(actor, memo.id))


def test_lost_create_reply_remains_unresolved_across_restart(handoff):
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    def lost(body):
        calls.append(body)
        raise OSError('lost receiver reply')
    intake.creator = lost
    assert request(intake, receipt['envelope'])[0] == 503
    intake.store = Store(intake.store.path)
    assert request(intake, receipt['envelope']) == (409, {'error': 'approval_create_unresolved'})
    assert len(calls) == 1
    assert app.outcomes(actor, memo.id, 1)['review']['intake_state'] == 'creating'


def test_schema_four_upgrade_preserves_evidence_and_fences_old_readers(handoff):
    app, actor, memo, options, intake, calls = handoff
    store = app.repository.store
    before = app.events(actor, memo.id)
    # Reconstruct the prior schema's exact index/version boundary.
    with store._connection() as db:
        db.execute('DROP INDEX lifecycle_review_snapshot')
        db.execute('DROP INDEX lifecycle_review_revision')
        db.execute('PRAGMA user_version=4')
    migrated = Store(store.path)
    with migrated._connection() as db:
        assert db.execute('PRAGMA user_version').fetchone()[0] == 5
        assert db.execute("SELECT count(*) FROM sqlite_master WHERE name IN "
                          "('lifecycle_review_snapshot','lifecycle_review_revision')").fetchone()[0] == 2
    assert app.events(actor, memo.id) == before
    receipt = prepare(handoff)
    assert prepare(handoff) == receipt


def test_attribute_review_checks_live_requester_before_policy(handoff):
    from informed_decision.review import ReviewController, ReviewError
    from test_policy_client import human
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    assert request(intake, receipt['envelope'])[0] == 201
    stored = app.repository.store.memo(receipt['review_memo_id'])
    session = replace(human(), subject=stored.binding.principal.id)
    class Engine:
        def get_approval(self, approval_id):
            return {'binding': {'digest': stored.approval_binding_digest, 'actor': 'wrong-requester', 'human_control': True}}
    controller = ReviewController(app.repository.store, policy(), lambda session: Engine(), intake_attributes=True)
    with pytest.raises(ReviewError, match='binding_changed'):
        controller.open(session, stored.id)
    assert controller.policy.transport.calls == []


def test_intake_policy_guard_rechecks_retained_metadata(handoff):
    from informed_decision.policy import build_request
    from test_policy_client import human
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    assert request(intake, receipt['envelope'])[0] == 201
    store = app.repository.store
    review = store.memo(receipt['review_memo_id'])
    session = replace(human(), subject=review.binding.principal.id)
    request_body = build_request(session, review, 'read', 'v1', intake_context=store.intake_context(review))
    observation = policy().check(request_body)
    policy_id = store.observe_policy(observation)
    with store._transaction() as db:
        row = db.execute('SELECT metadata FROM intake_ledger WHERE memo_id=?', (review.id,)).fetchone()
        metadata = json.loads(row['metadata']); metadata['review_group'] = 'changed'
        db.execute('UPDATE intake_ledger SET metadata=? WHERE memo_id=?', (json.dumps(metadata), review.id))
    with pytest.raises(Conflict, match='policy observation'):
        store.present(review.id, principal_sub=session.subject, tenant=session.tenant,
                      principal_type=session.principal_type, policy_id=policy_id)


def test_retained_handoff_api_is_revision_scoped_and_reauthorizes(handoff):
    from informed_decision.lifecycle_http import LifecycleAPI
    from test_lifecycle_http import call
    app, actor, memo, options, intake, calls = handoff
    api = LifecycleAPI(app, lambda _: actor)
    path = '/v1/decisions/' + memo.id + '/review-requests'
    assert call(api, path, QUERY_STRING='version=1')[0] == 404
    receipt = prepare(handoff)
    assert call(api, path, QUERY_STRING='version=1')[:2] == (200, receipt)
    assert not calls
    for query in ('', 'version=0', 'version=1&extra=x', 'version=1&version=1'):
        assert call(api, path, QUERY_STRING=query)[0] == 400
    app.revise(actor, memo.next_version(brief='Next revision'), expected_version=1, summary='Refined', key='revise')
    assert call(api, path, QUERY_STRING='version=2')[0] == 404
    # Historical export is evidence, never permission to resubmit a stale request.
    assert call(api, path, QUERY_STRING='version=1')[:2] == (200, receipt)
    assert request(intake, receipt['envelope'])[0] == 409
    app.policy.transport.change = lambda data: data.update(effect='deny')
    assert call(api, path, query='version=1')[:2] == (403, {'error': 'policy_denied'})


def test_overview_uses_retained_intake_context_and_fails_closed_per_row(handoff):
    from informed_decision.review import ReviewController
    from informed_decision.approval_client import ApprovalEngineError
    from test_policy_client import human
    app, actor, memo, options, intake, calls = handoff
    receipt = prepare(handoff)
    assert request(intake, receipt['envelope'])[0] == 201
    store = app.repository.store
    stored = store.memo(receipt['review_memo_id'])
    session = replace(human(), subject=stored.binding.principal.id)
    class Engine:
        failure = False
        def get_approval(self, approval_id):
            if self.failure:
                raise ApprovalEngineError(503, 'unavailable')
            return {'status': 'requested', 'binding': {'digest': stored.approval_binding_digest,
                    'actor': 'informed-decision', 'human_control': True}}
    engine = Engine()
    controller = ReviewController(store, policy(), lambda session: engine, intake_attributes=True)
    [row] = controller.overview(session)
    assert row.question == memo.question
    sent = controller.policy.transport.calls[-1][3]
    assert sent['action'] == 'list'
    assert sent['context']['review_recipient'] == session.subject
    assert sent['context']['intake_submitter'] == 'state-hub'
    assert 'review_recipient' not in build_request(
        session, stored, 'read', 'v1', intake_context=store.intake_context(stored))['context']
    assert store.evidence() == []
    engine.failure = True
    [row] = controller.overview(session)
    assert row.group == 'unavailable' and row.question is None
    assert len(controller.policy.transport.calls) == 1


def test_real_discovery_policy_lists_only_recipient_and_keeps_fresh_read(handoff, tmp_path, signing_key):
    import os
    from pathlib import Path
    from informed_decision.policy import build_request
    from informed_decision.review import ReviewController, ReviewError
    from review_components import flex_component
    from test_policy_client import human
    binary, source = os.environ.get('INFD_FLEX_AUTH_BINARY'), os.environ.get('INFD_FLEX_AUTH_SOURCE')
    if not binary or not source:
        pytest.skip('actual owner discovery policy required')
    fixture = Path(source) / 'examples/informed-decision-lifecycle-review'
    if json.loads((fixture / 'pin.json').read_text())['version'] != 'v13':
        pytest.skip('owner discovery v13 required')
    app, actor, memo, options, intake, calls = handoff
    options['memo_class'] = 'decision-record'
    receipt = prepare(handoff)
    # Admit this class in the synthetic intake registry used by the fixture.
    registry = json.loads(intake.submitters_path.read_text())
    for value in registry.values():
        value['memo_classes'].append('decision-record')
    intake.submitters_path.write_text(json.dumps(registry))
    assert request(intake, receipt['envelope'])[0] == 201
    store = app.repository.store
    stored = store.memo(receipt['review_memo_id'])
    session = replace(human(), subject=stored.binding.principal.id,
                      assurance={**human().assurance, 'at': int(time.time()) - 3600})
    class Engine:
        def get_approval(self, approval_id):
            return {'status': 'requested', 'binding': {'digest': stored.approval_binding_digest,
                    'actor': 'informed-decision', 'human_control': True}}
    with flex_component(binary, tmp_path, signing_key, fixture=fixture) as (pdp, calls):
        controller = ReviewController(store, pdp, lambda session: Engine(), intake_attributes=True)
        [row] = controller.overview(session)
        assert row.question == memo.question and row.group == 'open'
        with pytest.raises(ReviewError, match='policy_denied'):
            controller.open(session, stored.id)
        req = build_request(session, stored, 'list', pdp.version, intake_context=store.intake_context(stored))
        req['context']['review_recipient'] = 'another-human'
        assert pdp.check(req).outcome == 'policy_denied'
        assert store.evidence() == []