from __future__ import annotations

import hashlib
import json
import unittest
from base64 import b64decode, urlsafe_b64encode
from dataclasses import dataclass
from typing import Any, cast

from services.runtime.ingestion_auth import ActorContext
from services.runtime.ingestion_files import FileIngestionError, FilePolicy, ProcessedSource
from services.runtime.ingestion_hub_client import HubClient
from services.runtime.ingestion_workflow import IngestionWorkflow

PROJECT_ID = "11111111-1111-4111-8111-111111111111"
PROPOSAL_ID = "22222222-2222-4222-8222-222222222222"
OTHER_PAGE_ID = "33333333-3333-4333-8333-333333333333"
PROJECT_PAGE_ID = "44444444-4444-4444-8444-444444444444"


def _git_blob_sha(text: str, *, sha256: bool = False) -> str:
    raw = text.encode("utf-8")
    blob = f"blob {len(raw)}\0".encode("ascii") + raw
    return (
        hashlib.sha256(blob).hexdigest()
        if sha256
        else hashlib.sha1(blob, usedforsecurity=False).hexdigest()
    )


@dataclass
class _Reply:
    data: dict[str, Any]
    etag: str | None = None
    status_code: int = 200
    correlation_id: str = "trace-1"

    @property
    def id(self) -> str | None:
        value = self.data.get("id")
        return str(value) if value is not None else None


class _Hub:
    def __init__(self) -> None:
        self.calls: list[tuple[str, str, dict[str, Any]]] = []
        self.uploads: list[tuple[str, Any, str | None, str, str | None]] = []
        self.etag = '"v1"'
        self.detail: dict[str, Any] = {
            "id": PROPOSAL_ID,
            "correlationId": "trace-1",
            "status": "Received",
            "manifest": json.dumps({"projectId": PROJECT_ID, "userManifest": "Quarterly runbook"}),
            "destination": {
                "pageId": None,
                "parentPageId": None,
                "title": "Quarterly restart guide",
            },
            "metadata": {"documentType": "Runbook", "documentStatus": "Draft"},
            "analysis": None,
            "reviewerComment": None,
            "retryCount": 0,
            "maxPublishRetries": 3,
            "isRetryEligible": False,
            "sources": [],
            "citations": [],
        }

    async def request_json(self, method: str, path: str, **kwargs: Any) -> _Reply:
        self.calls.append((method, path, kwargs))
        if method == "POST" and path == "/api/v1/ingestion-proposals":
            self.detail["manifest"] = kwargs["json_body"]["manifest"]
            return _Reply({"id": PROPOSAL_ID, "status": "Received"}, etag='"v1"', status_code=201)
        if method == "GET" and path in {
            "/api/v1/pages/roots?limit=2",
            "/api/v1/pages/roots?cursor=roots-next&limit=2",
        }:
            return _Reply(
                {
                    "items": [
                        {
                            "id": PROJECT_PAGE_ID,
                            "title": "Course",
                            "slug": "course",
                            "documentType": "Course",
                            "documentStatus": "Published",
                            "hasChildren": True,
                        }
                    ],
                    "nextCursor": "roots-next" if "cursor=" not in path else None,
                }
            )
        if method == "GET" and path == f"/api/v1/pages/{PROJECT_PAGE_ID}/children?limit=2":
            return _Reply(
                {
                    "items": [
                        {
                            "id": OTHER_PAGE_ID,
                            "parentPageId": PROJECT_PAGE_ID,
                            "title": "Module",
                            "slug": "module",
                            "documentType": "Module",
                            "documentStatus": "Published",
                            "hasChildren": False,
                        }
                    ],
                    "nextCursor": None,
                }
            )
        if method == "GET" and path == f"/api/v1/pages/{PROJECT_PAGE_ID}":
            return _Reply(
                {
                    "id": PROJECT_PAGE_ID,
                    "title": "Course",
                    "slug": "course",
                    "markdown": "# Course\n" + ("A" * 40),
                    "documentType": "Course",
                    "documentStatus": "Published",
                    "parentPageId": None,
                    "breadcrumb": [{"id": PROJECT_PAGE_ID, "title": "Course"}],
                },
                etag='"page-v7"',
            )
        if method == "GET" and path == "/api/v1/search?q=architecture&limit=10":
            return _Reply(
                {
                    "items": [
                        {
                            "id": PROJECT_PAGE_ID,
                            "title": "Project guide",
                            "slug": "guide",
                            "snippet": "Allowed",
                        },
                        {
                            "id": OTHER_PAGE_ID,
                            "title": "Other project",
                            "slug": "other",
                            "snippet": "Must be filtered",
                        },
                    ],
                    "nextCursor": None,
                }
            )
        if (
            method == "GET"
            and path
            == f"/api/v1/projects/{PROJECT_ID}/context?updates=10&decisions=10&documents=20"
        ):
            return _Reply(
                {
                    "project": {"id": PROJECT_ID, "name": "Project", "summary": "Scoped"},
                    "citations": [
                        {
                            "kind": "page",
                            "id": PROJECT_PAGE_ID,
                            "revision": 7,
                            "title": "Project guide",
                            "canonicalPath": f"/knowledge/{PROJECT_PAGE_ID}/guide",
                        }
                    ],
                    "truncation": {"truncated": False, "omittedCounts": {}},
                }
            )
        if method == "GET" and path == f"/api/v1/ingestion-proposals/{PROPOSAL_ID}":
            return _Reply(dict(self.detail), etag=self.etag)
        if method == "PATCH" and path == f"/api/v1/ingestion-proposals/{PROPOSAL_ID}":
            self.detail["analysis"] = kwargs["json_body"]["analysis"]
            self.detail["status"] = "ReadyForReview"
            self.etag = '"v3"'
            return _Reply(dict(self.detail), etag=self.etag)
        raise AssertionError(f"unexpected Hub request: {method} {path}")

    async def upload_source(self, proposal_id: str, source: Any, **kwargs: Any) -> _Reply:
        self.uploads.append(
            (
                proposal_id,
                source,
                kwargs.get("etag"),
                kwargs["idempotency_key"],
                kwargs["project_id"],
            )
        )
        self.detail["sources"].append(
            {
                "fileName": source.file_name,
                "mimeType": source.content_type,
                "byteSize": len(source.content),
                "digestSha256": "a" * 64,
                "scanResult": "Clean",
            }
        )
        self.etag = f'"v{len(self.uploads) + 1}"'
        return _Reply(dict(self.detail), etag=self.etag)


class _ActorAuthorizer:
    def __init__(self) -> None:
        self.calls: list[tuple[Any, str, str | None]] = []

    def __call__(self, actor: ActorContext, *, capability: str, project: str | None = None) -> None:
        self.calls.append((actor, capability, project))
        if capability not in actor.scopes or (
            project is not None and project not in actor.projects
        ):
            raise PermissionError("actor not authorized")


class IngestionWorkflowTests(unittest.IsolatedAsyncioTestCase):
    def setUp(self) -> None:
        self.actor = ActorContext(
            subject="user-1",
            projects=frozenset({PROJECT_ID}),
            scopes=frozenset({"knowledge:read", "ingestion:write"}),
            correlation_id="trace-1",
        )
        self.hub = _Hub()
        self.authorizer = _ActorAuthorizer()
        self.processed = (
            type(
                "ProcessedSourceFixture",
                (),
                {
                    "file_id": "file_1",
                    "file_name": "runbook.txt",
                    "mime_type": "text/plain",
                    "sha256": "a" * 64,
                    "text": "restart and verify",
                    "byte_size": 20,
                    "content": b"restart and verify\n",
                },
            )(),
        )

        async def acquire(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
            self.assertIs(actor, self.actor)
            self.assertIsNotNone(policy)
            self.assertEqual(files[0]["file_id"], "file_1")
            return self.processed

        self.workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            file_acquirer=acquire,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )

        async def scan_text_source(text: str, file_name: str, *, policy: Any) -> Any:
            raw = text.encode("utf-8")
            return ProcessedSource(
                "repo-source",
                file_name,
                "text/plain",
                hashlib.sha256(raw).hexdigest(),
                text,
                len(raw),
                raw,
            )

        self.text_source_scanner = scan_text_source

    async def test_analysis_upload_and_submit_use_canonical_scoped_review_flow(self) -> None:
        files = [
            {
                "download_url": "https://files.openai.com/temporary/secret-url",
                "file_id": "file_1",
                "mime_type": "text/plain",
                "file_name": "runbook.txt",
            }
        ]
        publication = {
            "title": "Quarterly restart guide",
            "markdown": "# Restart\nVerify service health.",
            "document_type": "Runbook",
            "document_status": "Draft",
            "mutation_kind": "CreatePage",
            "parent_page_id": None,
            "target_page_id": None,
            "expected_target_etag": None,
        }

        created = await self.workflow.analyze_knowledge_files(
            project_id=PROJECT_ID,
            files=files,
            manifest="User requested a restart guide.",
            publication=publication,
            actor=self.actor,
        )
        completed = await self.workflow.submit_ingestion_proposal(
            project_id=PROJECT_ID,
            proposal_id=PROPOSAL_ID,
            analysis="Source checked; this is a machine proposal for review.",
            actor=self.actor,
        )

        self.assertEqual(created["proposalId"], PROPOSAL_ID)
        self.assertEqual(created["status"], "Received")
        self.assertEqual(len(self.hub.uploads), 1)
        uploaded_proposal, source, upload_etag, _, upload_project = self.hub.uploads[0]
        self.assertEqual(uploaded_proposal, PROPOSAL_ID)
        self.assertEqual(source.content, b"restart and verify\n")
        self.assertEqual(upload_etag, '"v1"')
        self.assertEqual(upload_project, PROJECT_ID)

        create_method, create_path, create_kwargs = self.hub.calls[0]
        self.assertEqual((create_method, create_path), ("POST", "/api/v1/ingestion-proposals"))
        self.assertEqual(create_kwargs["project_id"], PROJECT_ID)
        manifest = json.loads(create_kwargs["json_body"]["manifest"])
        self.assertEqual(manifest["projectId"], PROJECT_ID)
        self.assertNotIn("download_url", create_kwargs["json_body"]["manifest"])
        self.assertEqual(
            create_kwargs["json_body"]["publication"]["mutationKind"],
            "CreatePage",
        )

        patch = next(call for call in self.hub.calls if call[0] == "PATCH")
        self.assertEqual(patch[1], f"/api/v1/ingestion-proposals/{PROPOSAL_ID}")
        self.assertEqual(
            patch[2]["json_body"],
            {"analysis": "Source checked; this is a machine proposal for review."},
        )
        self.assertEqual(patch[2]["etag"], '"v2"')
        self.assertEqual(patch[2]["project_id"], PROJECT_ID)
        self.assertEqual(completed["status"], "ReadyForReview")
        self.assertNotIn("restart and verify", json.dumps(created))
        self.assertNotIn("secret-url", json.dumps(created))
        self.assertFalse(
            any(
                path.endswith(("/submit", "/publish", "/approve", "/reject"))
                for _, path, _ in self.hub.calls
            )
        )

    async def test_search_filters_workspace_results_to_project_citations(self) -> None:
        result = await self.workflow.search_knowledge(
            project_id=PROJECT_ID,
            query="architecture",
            limit=10,
            actor=self.actor,
        )
        self.assertEqual([item["id"] for item in result["results"]], [PROJECT_PAGE_ID])
        self.assertEqual(result["results"][0]["citation"]["revision"], 7)
        self.assertEqual(
            [call[1] for call in self.hub.calls],
            [
                "/api/v1/search?q=architecture&limit=10",
                f"/api/v1/projects/{PROJECT_ID}/context?updates=10&decisions=10&documents=20",
            ],
        )

    async def test_projectless_workspace_flow_is_opted_in_and_keeps_project_context_separate(
        self,
    ) -> None:
        trusted = ActorContext(
            subject="trusted-client",
            projects=frozenset({PROJECT_ID}),
            scopes=frozenset({"knowledge:read", "ingestion:write"}),
            correlation_id="trace-1",
            connector_project="knowledge-hub",
            connector_environment="prod",
        )

        async def acquire_workspace(
            files: Any, *, actor: ActorContext, policy: Any
        ) -> tuple[Any, ...]:
            self.assertIs(actor, trusted)
            return self.processed

        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            file_acquirer=acquire_workspace,
            workspace_read_enabled=True,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )
        with self.assertRaises(PermissionError):
            await self.workflow.search_knowledge(
                project_id=None, query="architecture", actor=trusted
            )
        self.assertEqual(self.hub.calls, [])
        search = await workflow.search_knowledge(
            project_id=None, query="architecture", actor=trusted
        )
        self.assertEqual(
            [item["id"] for item in search["results"]], [PROJECT_PAGE_ID, OTHER_PAGE_ID]
        )
        self.assertEqual(
            [call[1] for call in self.hub.calls], ["/api/v1/search?q=architecture&limit=10"]
        )
        self.assertIsNone(search["projectId"])

        files = [{"download_url": "https://files.openai.com/file", "file_id": "file_1"}]
        extracted = await workflow.extract_knowledge_files(
            project_id=None, files=files, actor=trusted
        )
        self.assertEqual(extracted["page"]["text"], "restart and verify")
        self.assertEqual(len(self.hub.calls), 1)
        created = await workflow.analyze_knowledge_files(
            project_id=None,
            files=files,
            manifest="Review the course content.",
            publication={
                "title": "Course guide",
                "markdown": "# Course guide",
                "document_type": "Course",
                "document_status": "Draft",
                "mutation_kind": "CreatePage",
                "parent_page_id": None,
            },
            actor=trusted,
        )
        self.assertEqual(created["proposalId"], PROPOSAL_ID)
        manifest = json.loads(self.hub.detail["manifest"])
        self.assertEqual(manifest["workspace"], "knowledge-hub")
        self.assertNotIn("projectId", manifest)
        self.assertIsNone(self.hub.uploads[0][-1])
        submitted = await workflow.submit_ingestion_proposal(
            project_id=None,
            proposal_id=PROPOSAL_ID,
            analysis="Ready for human review.",
            actor=trusted,
        )
        status = await workflow.get_ingestion_status(
            project_id=None, proposal_id=PROPOSAL_ID, actor=trusted
        )
        self.assertEqual(submitted["status"], "ReadyForReview")
        self.assertEqual(status["status"], "ReadyForReview")

        wrong = ActorContext(
            subject="untrusted-client",
            projects=frozenset({PROJECT_ID}),
            scopes=frozenset({"knowledge:read", "ingestion:write"}),
            correlation_id="trace-2",
        )
        calls_before = len(self.hub.calls)
        with self.assertRaises(PermissionError):
            await workflow.get_ingestion_status(
                project_id=None, proposal_id=PROPOSAL_ID, actor=wrong
            )
        with self.assertRaises(PermissionError):
            await workflow.search_knowledge(project_id=OTHER_PAGE_ID, query="course", actor=trusted)
        self.assertEqual(len(self.hub.calls), calls_before)

    async def test_project_scope_denial_happens_before_any_hub_request(self) -> None:
        actor = ActorContext(
            subject="user-2",
            projects=frozenset(),
            scopes=frozenset({"knowledge:read"}),
            correlation_id="trace-2",
        )

        with self.assertRaises(PermissionError):
            await self.workflow.get_knowledge_context(project_id=PROJECT_ID, actor=actor)

        self.assertEqual(self.hub.calls, [])

    async def test_rejected_file_creates_no_proposal_shell(self) -> None:
        async def reject_files(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
            raise ValueError("private download URL and extracted body must never be returned")

        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            file_acquirer=reject_files,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )
        with self.assertRaises(ValueError):
            await workflow.analyze_knowledge_files(
                project_id=PROJECT_ID,
                files=[{"download_url": "https://invalid.test/file", "file_id": "file_1"}],
                manifest="Fixture",
                publication={
                    "title": "Guide",
                    "markdown": "# Guide",
                    "document_type": "Runbook",
                    "document_status": "Draft",
                    "mutation_kind": "CreatePage",
                },
                actor=self.actor,
            )
        self.assertEqual(self.hub.calls, [])

    async def test_no_extractable_text_cannot_create_or_complete_proposal(self) -> None:
        async def image_only(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
            source = self.processed[0]
            return (
                ProcessedSource(
                    source.file_id,
                    "image-only.pdf",
                    "application/pdf",
                    source.sha256,
                    "",
                    source.byte_size,
                    source.content,
                ),
            )

        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            file_acquirer=image_only,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )
        files = [{"download_url": "https://files.openai.com/image-only.pdf", "file_id": "file_1"}]
        with self.assertRaises(FileIngestionError):
            await workflow.extract_knowledge_files(
                project_id=PROJECT_ID, files=files, actor=self.actor
            )
        with self.assertRaises(FileIngestionError):
            await workflow.analyze_knowledge_files(
                project_id=PROJECT_ID,
                files=files,
                manifest="Image only",
                publication={
                    "title": "Image",
                    "markdown": "# Image",
                    "document_type": "Course",
                    "document_status": "Draft",
                    "mutation_kind": "CreatePage",
                    "parent_page_id": None,
                },
                actor=self.actor,
            )
        self.assertEqual(self.hub.calls, [])
        self.assertEqual(self.hub.uploads, [])

    async def test_attachment_extraction_returns_real_text_with_per_file_coverage_and_no_proposal(
        self,
    ) -> None:
        text = "DISTINCTIVE-BEGIN\n" + ("line-knowledge-content\n" * 900) + "DISTINCTIVE-END"
        processed = (
            type(
                "ProcessedSourceFixture",
                (),
                {
                    "file_id": "file_1",
                    "file_name": "guide.txt",
                    "mime_type": "text/plain",
                    "sha256": "b" * 64,
                    "text": text,
                    "byte_size": len(text.encode()),
                    "content": text.encode(),
                },
            )(),
        )

        async def acquire(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
            return processed

        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            file_acquirer=acquire,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )
        file_input = [{"download_url": "https://files.openai.com/file", "file_id": "file_1"}]
        first = await workflow.extract_knowledge_files(
            project_id=PROJECT_ID, files=file_input, actor=self.actor
        )

        self.assertIn("DISTINCTIVE-BEGIN", first["page"]["text"])
        self.assertEqual(first["page"]["sha256"], "b" * 64)
        self.assertEqual(first["page"]["startOffset"], 0)
        self.assertFalse(first["page"]["complete"])
        self.assertEqual(first["fileCoverage"][0]["coveredChars"], len(first["page"]["text"]))
        self.assertIsNotNone(first["nextCursor"])
        cursor = first["nextCursor"]
        encoded = cursor[3:]
        raw = b64decode(encoded + "=" * (-len(encoded) % 4), altchars=b"-_", validate=True)
        payload = json.loads(raw[:-32])
        payload["offset"] = len(text) - 1
        forged = "v2:" + urlsafe_b64encode(
            json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + raw[-32:]
        ).decode().rstrip("=")
        with self.assertRaises(ValueError):
            await workflow.extract_knowledge_files(
                project_id=PROJECT_ID, files=file_input, page_cursor=forged, actor=self.actor
            )
        other_actor = ActorContext(
            subject="other-user",
            projects=frozenset({PROJECT_ID}),
            scopes=frozenset({"ingestion:write", "knowledge:read"}),
            correlation_id="trace-other",
        )
        with self.assertRaises(ValueError):
            await workflow.extract_knowledge_files(
                project_id=PROJECT_ID, files=file_input, page_cursor=cursor, actor=other_actor
            )
        second = await workflow.extract_knowledge_files(
            project_id=PROJECT_ID,
            files=file_input,
            page_cursor=first["nextCursor"],
            actor=self.actor,
        )

        self.assertIn("DISTINCTIVE-END", second["page"]["text"])
        self.assertTrue(second["page"]["complete"])
        self.assertTrue(second["allFilesComplete"])
        self.assertIsNone(second["nextCursor"])
        self.assertEqual(self.hub.calls, [])
        self.assertEqual(self.hub.uploads, [])

    async def test_workspace_read_is_default_off_and_checks_exact_trusted_identity(self) -> None:
        with self.assertRaises(PermissionError):
            await self.workflow.get_workspace_roots(actor=self.actor, limit=2)
        self.assertEqual(self.hub.calls, [])

        workspace_actor = ActorContext(
            subject="trusted-client",
            projects=frozenset({PROJECT_ID}),
            scopes=frozenset({"knowledge:read", "ingestion:write"}),
            correlation_id="trace-1",
            connector_project="knowledge-hub",
            connector_environment="prod",
        )
        enabled = IngestionWorkflow(
            cast(HubClient, self.hub),
            workspace_read_enabled=True,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )
        roots = await enabled.get_workspace_roots(actor=workspace_actor, limit=2)
        children = await enabled.get_workspace_children(
            parent_page_id=PROJECT_PAGE_ID, actor=workspace_actor, limit=2
        )
        page = await enabled.get_workspace_page(
            page_id=PROJECT_PAGE_ID, actor=workspace_actor, max_chars=10
        )

        self.assertEqual(roots["items"][0]["title"], "Course")
        self.assertEqual(roots["nextCursor"], "roots-next")
        self.assertEqual(children["items"][0]["parentPageId"], PROJECT_PAGE_ID)
        self.assertEqual(page["etag"], '"page-v7"')
        self.assertEqual(page["markdown"], "# Course\nA")
        self.assertEqual(page["nextOffset"], 10)
        self.assertFalse(page["complete"])

        for marker in ("other-hub", "staging"):
            wrong_actor = ActorContext(
                subject="trusted-client",
                projects=frozenset({PROJECT_ID}),
                scopes=frozenset({"knowledge:read"}),
                correlation_id="trace-1",
                connector_project=marker,
                connector_environment="prod" if marker == "other-hub" else "staging",
            )
            with self.subTest(marker=marker), self.assertRaises(PermissionError):
                await enabled.get_workspace_roots(actor=wrong_actor, limit=2)

    async def test_workspace_page_pagination_requires_a_stable_etag_and_update_uses_it(
        self,
    ) -> None:
        workspace_actor = ActorContext(
            subject="trusted-client",
            projects=frozenset({PROJECT_ID}),
            scopes=frozenset({"knowledge:read", "ingestion:write"}),
            correlation_id="trace-1",
            connector_project="knowledge-hub",
            connector_environment="prod",
        )
        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            workspace_read_enabled=True,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )
        with self.assertRaises(ValueError):
            await workflow.get_workspace_page(
                page_id=PROJECT_PAGE_ID,
                actor=workspace_actor,
                offset_chars=10,
                max_chars=10,
            )
        page = await workflow.get_workspace_page(
            page_id=PROJECT_PAGE_ID,
            actor=workspace_actor,
            offset_chars=10,
            max_chars=10,
            expected_etag='"page-v7"',
        )
        self.assertEqual(page["markdown"], "AAAAAAAAAA")
        self.assertEqual(page["etag"], '"page-v7"')

        update = self.workflow._validate_publication(
            {
                "title": "Course",
                "markdown": "# Revised course",
                "document_type": "Course",
                "document_status": "Draft",
                "mutation_kind": "UpdatePage",
                "target_page_id": PROJECT_PAGE_ID,
                "expected_target_etag": page["etag"],
            }
        )
        self.assertEqual(update["expectedTargetETag"], page["etag"])

    async def test_text_source_proposal_keeps_caller_provenance_and_full_publication(self) -> None:
        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
            text_source_scanner=self.text_source_scanner,
        )
        result = await workflow.create_text_source_proposal(
            project_id=PROJECT_ID,
            source_file_name="guide.md",
            source_text="# Source guide\nVerified instructions.",
            source_complete=True,
            source_uri="itworx-git://ludarium/app/docs/guide.md",
            source_ref="main",
            source_commit="a" * 40,
            source_sha=_git_blob_sha("# Source guide\nVerified instructions."),
            manifest="Propose a reviewed guide.",
            publication={
                "title": "Reviewed guide",
                "markdown": "# Reviewed guide\nCleaned and current.",
                "document_type": "Runbook",
                "document_status": "Draft",
                "mutation_kind": "CreatePage",
                "parent_page_id": None,
            },
            actor=self.actor,
        )

        self.assertEqual(result["proposalId"], PROPOSAL_ID)
        create = next(call for call in self.hub.calls if call[0] == "POST")
        body = create[2]["json_body"]
        self.assertEqual(body["publication"]["title"], "Reviewed guide")
        manifest = json.loads(body["manifest"])
        self.assertEqual(
            manifest["sourceProvenance"][0]["sourceUri"], "itworx-git://ludarium/app/docs/guide.md"
        )
        self.assertEqual(
            manifest["sourceProvenance"][0]["sourceSha"],
            _git_blob_sha("# Source guide\nVerified instructions."),
        )
        self.assertFalse(manifest["sourceProvenance"][0]["verified"])
        self.assertEqual(self.hub.uploads[0][1].content, b"# Source guide\nVerified instructions.")

    async def test_projectless_text_source_proposal_requires_trusted_workspace(self) -> None:
        trusted = ActorContext(
            subject="trusted-client",
            projects=frozenset(),
            scopes=frozenset({"knowledge:read", "ingestion:write"}),
            correlation_id="trace-workspace",
            connector_project="knowledge-hub",
            connector_environment="prod",
        )
        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            text_source_scanner=self.text_source_scanner,
            workspace_read_enabled=True,
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
        )
        result = await workflow.create_text_source_proposal(
            project_id=None,
            source_file_name="guide.md",
            source_text="# Complete source",
            source_complete=True,
            source_uri="itworx-git://school/courses/docs/guide.md",
            source_ref="main",
            source_commit="a" * 40,
            source_sha=_git_blob_sha("# Complete source", sha256=True),
            manifest="Course source",
            publication={
                "title": "Guide",
                "markdown": "# Guide",
                "document_type": "Course",
                "document_status": "Draft",
                "mutation_kind": "CreatePage",
                "parent_page_id": None,
            },
            actor=trusted,
        )
        self.assertEqual(result["proposalId"], PROPOSAL_ID)
        manifest = json.loads(self.hub.detail["manifest"])
        self.assertEqual(manifest["workspace"], "knowledge-hub")
        self.assertNotIn("projectId", manifest)
        self.assertFalse(manifest["sourceProvenance"][0]["verified"])
        self.assertIsNone(self.hub.uploads[0][-1])

    async def test_text_source_rejects_incomplete_or_invalid_provenance_before_proposal(
        self,
    ) -> None:
        workflow = IngestionWorkflow(
            cast(HubClient, self.hub),
            file_policy=cast(FilePolicy, object()),
            actor_authorizer=self.authorizer,
            cursor_signing_key=b"c" * 32,
            text_source_scanner=self.text_source_scanner,
        )
        arguments = {
            "project_id": PROJECT_ID,
            "source_file_name": "guide.md",
            "source_text": "# Source guide",
            "source_complete": False,
            "source_uri": "itworx-git://ludarium/app/docs/guide.md",
            "source_ref": "main",
            "source_commit": "a" * 40,
            "source_sha": "b" * 40,
            "manifest": "Fixture",
            "publication": {
                "title": "Guide",
                "markdown": "# Guide",
                "document_type": "Runbook",
                "document_status": "Draft",
                "mutation_kind": "CreatePage",
                "parent_page_id": None,
            },
            "actor": self.actor,
        }
        with self.assertRaises(ValueError):
            await workflow.create_text_source_proposal(**arguments)
        self.assertEqual(self.hub.calls, [])
        arguments["source_complete"] = True
        with self.assertRaisesRegex(ValueError, "Git blob SHA"):
            await workflow.create_text_source_proposal(**arguments)
        self.assertEqual(self.hub.calls, [])


if __name__ == "__main__":
    unittest.main()