state-hub/tests/test_forge_projection_qualifiers.py

"""File-derived qualifiers survive normal receiving, including source clears."""
from datetime import date
from pathlib import Path

import pytest

from api.services import forge_projection as fp
from tests.conftest import create_test_domain, create_test_repo


def _source(tmp_path: Path, *, workplan_fields='', task_fields='', extra_fields=None):
    root = tmp_path / 'projection-test'
    path = root / 'workplans/TEST-WP-0001.md'
    path.parent.mkdir(parents=True, exist_ok=True)
    body = (
        '---\nid: TEST-WP-0001\ntype: workplan\ntitle: Receiving test\nstatus: active\n'
        + workplan_fields + '---\n\n## Gate\n\n```task\nid: TEST-WP-0001-T01\n'
        'status: wait\npriority: high\n' + task_fields + '```\n\nRetain the owner gate.\n'
    )
    if extra_fields is not None:
        body += '\n## Second gate\n\n```task\nid: TEST-WP-0001-T02\nstatus: wait\n' + extra_fields + '```\n'
    path.write_text(body)
    return root, path


@pytest.mark.parametrize('value', ['2099-01-31', '"2099-01-31"'])
def test_derivation_and_json_retain_normalized_metadata(tmp_path, value):
    root, _ = _source(tmp_path,
        workplan_fields='flavor: " RESIDUAL "\nflavor_promotion_reason: " Risk "\nflavor_promoted_from: " Planning "\n',
        task_fields='flavor: " IMPLEMENTATION "\nneeds_human: "false"\ndecision_id: " decision-1 "\n'
                    'blocked_on: " message-from:provider "\nwait_until: ' + value + '\n')
    derived = fp.derive_from_checkout(root, 'projection-test', 'a' * 40)
    workplan, task = derived.workplans[0], derived.workplans[0].tasks[0]
    assert (workplan.flavor, workplan.flavor_promotion_reason, workplan.flavor_promoted_from) == ('residual', 'risk', 'planning')
    assert task.needs_human is False
    assert task.flavor == 'implementation'
    assert task.decision_id == 'decision-1'
    assert task.blocked_on == 'message-from:provider'
    assert task.wait_until == date(2099, 1, 31)
    payload = derived.to_dict()['workplans'][0]
    assert payload['flavor'] == 'residual'
    assert payload['flavor_promotion_reason'] == 'risk'
    assert payload['flavor_promoted_from'] == 'planning'
    assert payload['tasks'][0]['wait_until'] == '2099-01-31'


@pytest.mark.parametrize('field', [
    'wait_until: not-a-date\n', 'wait_until: 2099-02-31\n', 'wait_until: [2099-01-31]\n',
    'wait_until: 2099-01-31T12:30:00Z\n', 'wait_until: "2099-01-31T00:00:00Z"\n',
    'needs_human: "perhaps"\n', 'decision_id: [decision-1]\n',
    'blocked_on: {owner: provider}\n', 'flavor: invented\n',
    'decision_id: [unclosed\n',
])
def test_invalid_task_metadata_refuses_derivation(tmp_path, field):
    root, _ = _source(tmp_path, task_fields=field)
    with pytest.raises(fp.ForgeDeriveError):
        fp.derive_from_checkout(root, 'projection-test', 'a' * 40)


@pytest.mark.parametrize('field', ['flavor: invented\n', 'flavor_promoted_from: invented\n', 'flavor_promotion_reason: invented\n'])
def test_invalid_workplan_flavor_metadata_refuses_derivation(tmp_path, field):
    root, _ = _source(tmp_path, workplan_fields=field)
    with pytest.raises(fp.ForgeDeriveError):
        fp.derive_from_checkout(root, 'projection-test', 'a' * 40)


def test_dry_run_diff_exposes_qualifier_and_flavor_drift(tmp_path):
    root, _ = _source(tmp_path,
        workplan_fields='flavor: residual\nflavor_promotion_reason: demand\nflavor_promoted_from: planning\n',
        task_fields='flavor: implementation\nneeds_human: true\nblocking_reason: Owner must review.\n'
                    'decision_id: decision-1\nblocked_on: message-from:provider\nwait_until: 2099-01-31\n')
    derived = fp.derive_from_checkout(root, 'projection-test', 'a' * 40)
    workplan, task = derived.workplans[0], derived.workplans[0].tasks[0]
    hub_workplan = {'id': workplan.uuid, 'slug': workplan.record_id.lower(), 'status': 'active',
                    'backing_relative_path': workplan.relative_path}
    hub_task = {'id': task.uuid, 'record_id': task.record_id, 'status': 'wait', 'needs_human': False}
    diff = fp.diff_against_hub(derived, [hub_workplan], {workplan.uuid: [hub_task]})
    changes = {item['kind']: item['changed'] for item in diff.differing}
    assert set(changes['workplan']) == {'flavor', 'flavor_promotion_reason', 'flavor_promoted_from'}
    assert set(changes['task']) == {'flavor', 'needs_human', 'blocking_reason', 'intervention_note', 'decision_id', 'blocked_on', 'wait_until'}
    assert changes['task']['wait_until']['forge'] == '2099-01-31'
    # Diff is read-only, including the synthetic snapshot inputs.
    assert hub_task['needs_human'] is False
    assert 'flavor' not in hub_workplan


@pytest.mark.asyncio
async def test_real_receiving_creates_updates_and_clears_source_metadata(client, monkeypatch, tmp_path):
    await create_test_domain(client)
    repo = await create_test_repo(client, slug='projection-test')
    from api.config import settings
    from api.routers import work_record_projection as route
    monkeypatch.setattr(settings, 'state_hub_instance_role', 'primary')
    monkeypatch.setattr(settings, 'state_hub_instance_label', 'railiance01')
    root, _ = _source(tmp_path,
        workplan_fields='flavor: residual\n',
        task_fields='flavor: residual\nneeds_human: true\nblocking_reason: Founder must review.\n'
                    'decision_id: decision-1\nblocked_on: message-from:provider\nwait_until: 2099-01-31\n')
    revision = ['a' * 40]
    monkeypatch.setattr(route, 'derive_from_forge', lambda slug: fp.derive_from_checkout(root, slug, revision[0]))
    endpoint = '/repos/projection-test/work-record-projection/reconcile'

    async def reconcile(key):
        response = await client.post(endpoint, json={'expected_commit': revision[0]}, headers={'Idempotency-Key': key})
        assert response.status_code == 200, response.text
        return response.json()['outcome']

    async def snapshot():
        response = await client.get('/repos/projection-test/work-record-projection/snapshot')
        assert response.status_code == 200
        return response.json()

    first = await reconcile('qualifiers:create')
    assert first['status'] == 'applied'
    data = await snapshot()
    workplan = data['workplans'][0]
    task = data['tasks'][0]
    assert workplan['flavor'] == task['flavor'] == 'residual'
    assert task['needs_human'] is True
    assert task['intervention_note'] == 'Founder must review.'
    assert task['decision_id'] == 'decision-1'
    assert task['blocked_on'] == 'message-from:provider'
    assert task['wait_until'] == '2099-01-31'
    hidden = (await client.get('/workplans/', params={'repo_id': repo['id'], 'include_residuals': 'false'})).json()
    assert hidden == []
    summary = (await client.get('/state/summary')).json()
    assert workplan['id'] not in {row['id'] for row in summary['open_workplans']}

    # Existing workplan, existing task, and a new task on that existing parent.
    _source(tmp_path,
        workplan_fields='flavor: implementation\nflavor_promotion_reason: risk\nflavor_promoted_from: residual\n',
        task_fields='flavor: extension\nneeds_human: true\nblocking_reason: Founder must review.\n'
                    'decision_id: decision-2\nblocked_on: message-from:other\nwait_until: 2099-02-01\n',
        extra_fields='flavor: residual\ndecision_id: decision-3\nblocked_on: message-from:new-owner\nwait_until: 2099-03-01\n')
    revision[0] = 'b' * 40
    assert (await reconcile('qualifiers:update'))['status'] == 'applied'
    updated = await snapshot()
    workplan = updated['workplans'][0]
    tasks = {row['record_id']: row for row in updated['tasks']}
    assert (workplan['flavor'], workplan['flavor_promotion_reason'], workplan['flavor_promoted_from']) == ('implementation', 'risk', 'residual')
    task = tasks['TEST-WP-0001-T01']
    assert (task['flavor'], task['decision_id'], task['blocked_on'], task['wait_until']) == ('extension', 'decision-2', 'message-from:other', '2099-02-01')
    second = tasks['TEST-WP-0001-T02']
    assert (second['flavor'], second['decision_id'], second['blocked_on'], second['wait_until']) == ('residual', 'decision-3', 'message-from:new-owner', '2099-03-01')
    groups = {workplan['id']: list(tasks.values())}
    diff = fp.diff_against_hub(fp.derive_from_checkout(root, 'projection-test', revision[0]), [workplan], groups)
    assert diff.missing == diff.stale == diff.differing == []

    # Omitting qualifiers clears them; explicit null also clears. Quoted false
    # clears the human flag. Source identities survive every update.
    _source(tmp_path, task_fields='needs_human: "false"\ndecision_id: null\n', extra_fields='wait_until: null\n')
    revision[0] = 'c' * 40
    assert (await reconcile('qualifiers:clear'))['status'] == 'applied'
    cleared = await snapshot()
    assert cleared['workplans'][0]['id'] == workplan['id']
    for name in ('flavor', 'flavor_promotion_reason', 'flavor_promoted_from'):
        assert cleared['workplans'][0][name] is None
    assert {row['id'] for row in cleared['tasks']} == {row['id'] for row in updated['tasks']}
    for row in cleared['tasks']:
        assert row['needs_human'] is False
        for name in ('flavor', 'decision_id', 'blocked_on', 'wait_until', 'blocking_reason', 'intervention_note'):
            assert row[name] is None
    assert (await reconcile('qualifiers:independent-rederivation'))['status'] == 'noop'

    # Invalid source dates fail before receiving can replace/cancel stored gates.
    _source(tmp_path, task_fields='wait_until: 2099-02-31\n', extra_fields='')
    revision[0] = 'd' * 40
    invalid = await client.post(endpoint, json={'expected_commit': revision[0]}, headers={'Idempotency-Key': 'qualifiers:invalid'})
    assert invalid.status_code == 502
    unchanged = await snapshot()
    assert unchanged['workplans'] == cleared['workplans']
    assert unchanged['tasks'] == cleared['tasks']