diff --git a/application/api/user/sources/routes.py b/application/api/user/sources/routes.py index 114464bc..a8449bc2 100644 --- a/application/api/user/sources/routes.py +++ b/application/api/user/sources/routes.py @@ -10,7 +10,12 @@ from pydantic import ValidationError from application.agents.tools.path_utils import validate_tool_path from application.api import api -from application.api.user.tasks import reembed_wiki_page, reingest_source_task, sync_source +from application.api.user.tasks import ( + convert_source_to_wiki, + reembed_wiki_page, + reingest_source_task, + sync_source, +) from application.api.user.team_sharing import ( can_access, effective_write_owner, @@ -23,6 +28,7 @@ from application.storage.db.repositories.ingest_chunk_progress import ( ) from application.storage.db.repositories.sources import SourcesRepository from application.storage.db.repositories.wiki_pages import ( + WikiPageConflict, WikiPagesRepository, _content_hash, rebuild_wiki_directory_structure, @@ -624,6 +630,26 @@ class SourceConfigResource(Resource): # Ingest-time fields (config.chunking) only take effect after a # re-ingest (D8); compare against the current config to decide. current_config = SourceConfig.parse(doc.get("config")) + # kind changes route exclusively through /wiki/convert. An + # explicit kind in the body that differs from the persisted + # kind is rejected (either direction); otherwise the current + # kind is preserved so a partial edit omitting it (the model + # defaults to "classic") never demotes a wiki. + if "kind" in body and body["kind"] != current_config.kind: + return make_response( + jsonify( + { + "success": False, + "message": ( + "Change wiki mode via " + "POST /api/sources//wiki/convert, " + "not the config endpoint." + ), + } + ), + 400, + ) + new_config.kind = current_config.kind requires_reingest = ( new_config.chunking.model_dump() != current_config.chunking.model_dump() @@ -670,6 +696,7 @@ def _wiki_page_node(page): "token_count": page.get("token_count") or 0, "version": page.get("version"), "embed_status": page.get("embed_status"), + "updated_by": page.get("updated_by"), "updated_at": ( page["updated_at"].isoformat() if page.get("updated_at") is not None @@ -679,6 +706,18 @@ def _wiki_page_node(page): } +_wiki_page_edit_model = sources_ns.model( + "WikiPageEditModel", + { + "path": fields.String(required=True, description="Page path"), + "content": fields.String(required=True, description="Markdown content"), + "expected_version": fields.Integer( + description="Optimistic-lock version from the last read" + ), + }, +) + + @sources_ns.route("/sources/wiki") class CreateWikiSource(Resource): create_wiki_model = api.model( @@ -838,3 +877,208 @@ class WikiPage(Resource): node = _wiki_page_node(page) node["content"] = page.get("content") or "" return make_response(jsonify({"success": True, "page": node}), 200) + + @api.expect(_wiki_page_edit_model) + @api.doc( + description="Create or overwrite a wiki page (human edit). Write access " + "required; a stale expected_version returns 409." + ) + def put(self, source_id): + decoded_token = request.decoded_token + if not decoded_token: + return make_response(jsonify({"success": False}), 401) + user = decoded_token.get("sub") + data = request.get_json(silent=True) or {} + missing_fields = check_required_fields(data, ["path", "content"]) + if missing_fields: + return missing_fields + path = validate_tool_path(data["path"]) + if not path or path == "/" or path.endswith("/"): + return make_response( + jsonify({"success": False, "message": "Invalid page path"}), 400 + ) + content = data["content"] + expected_version = data.get("expected_version") + try: + with db_session() as conn: + owner = effective_write_owner(conn, "source", source_id, user) + if not owner: + return make_response( + jsonify( + {"success": False, "message": "Source not accessible"} + ), + 403, + ) + doc = SourcesRepository(conn).get_any(source_id, owner) + if doc is None: + return make_response( + jsonify({"success": False, "message": "Source not found"}), + 404, + ) + resolved_source_id = str(doc["id"]) + try: + page = WikiPagesRepository(conn).upsert( + resolved_source_id, + path, + content, + updated_by=user, + expected_version=expected_version, + ) + except WikiPageConflict: + return make_response( + jsonify( + { + "success": False, + "message": ( + "Page changed since you loaded it. " + "Reload and reapply your edit." + ), + } + ), + 409, + ) + rebuild_wiki_directory_structure(conn, resolved_source_id, owner) + except Exception as err: + current_app.logger.error( + f"Error editing wiki page for {source_id}: {err}", exc_info=True + ) + return make_response(jsonify({"success": False}), 400) + try: + reembed_wiki_page.delay( + resolved_source_id, + path, + _content_hash(content), + user=owner, + idempotency_key=( + f"reembed-wiki:{resolved_source_id}:{path}:" + f"{_content_hash(content)}" + ), + ) + except Exception as err: + current_app.logger.error( + f"Error enqueuing wiki re-embed for {source_id}: {err}", + exc_info=True, + ) + return make_response( + jsonify({"success": True, "page": _wiki_page_node(page)}), 200 + ) + + +def _source_is_blank(doc): + """True when a source has no ingested files to convert into pages.""" + structure = doc.get("directory_structure") or {} + if isinstance(structure, str): + try: + structure = json.loads(structure) + except Exception: + structure = {} + return not bool(structure) + + +def _enable_wiki_inline(conn, doc, owner): + """Flip a blank source to wiki mode + agentic exposure with no task.""" + cfg = SourceConfig.parse(doc.get("config")) + SourcesRepository(conn).update( + str(doc["id"]), owner, {"config": cfg.wiki_enabled()} + ) + + +@sources_ns.route("/sources//wiki/convert") +class ConvertSourceToWiki(Resource): + @api.doc( + description="Convert an ingested source into a wiki. A blank source is " + "enabled inline; a source with files runs a conversion task (poll its " + "status for the per-file summary). Write access required (owner/editor)." + ) + def post(self, source_id): + decoded_token = request.decoded_token + if not decoded_token: + return make_response(jsonify({"success": False}), 401) + user = decoded_token.get("sub") + try: + with db_session() as conn: + owner = effective_write_owner(conn, "source", source_id, user) + if not owner: + return make_response( + jsonify( + {"success": False, "message": "Source not accessible"} + ), + 403, + ) + doc = SourcesRepository(conn).get_any(source_id, owner) + if doc is None: + return make_response( + jsonify({"success": False, "message": "Source not found"}), + 404, + ) + resolved_source_id = str(doc["id"]) + # A mid-ingest source has an incomplete directory structure, so + # it could be mis-detected as blank and wrongly enabled inline. + # Reject while an embed is still in flight (mirrors the + # "processing" ingest status the sources list derives). + progress = IngestChunkProgressRepository(conn).get_progress( + resolved_source_id + ) + if ( + progress + and progress.get("status") != "stalled" + and (progress.get("embedded_chunks") or 0) + < (progress.get("total_chunks") or 0) + ): + return make_response( + jsonify( + { + "success": False, + "message": ( + "Source is still ingesting; wait for it to " + "finish before converting to a wiki." + ), + } + ), + 409, + ) + cfg = SourceConfig.parse(doc.get("config")) + if cfg.kind == "wiki": + return make_response( + jsonify( + { + "success": True, + "converted": False, + "enabled": True, + } + ), + 200, + ) + if _source_is_blank(doc): + _enable_wiki_inline(conn, doc, owner) + return make_response( + jsonify( + { + "success": True, + "converted": False, + "enabled": True, + } + ), + 200, + ) + except Exception as err: + current_app.logger.error( + f"Error preparing wiki conversion for {source_id}: {err}", + exc_info=True, + ) + return make_response(jsonify({"success": False}), 400) + try: + task = convert_source_to_wiki.delay( + source_id=resolved_source_id, + user=owner, + idempotency_key=f"convert-wiki:{resolved_source_id}", + ) + except Exception as err: + current_app.logger.error( + f"Error starting wiki conversion for {source_id}: {err}", + exc_info=True, + ) + return make_response(jsonify({"success": False}), 400) + return make_response( + jsonify({"success": True, "task_id": task.id}), 200 + ) diff --git a/application/api/user/tasks.py b/application/api/user/tasks.py index 9eed3202..4de4bc83 100644 --- a/application/api/user/tasks.py +++ b/application/api/user/tasks.py @@ -128,6 +128,15 @@ def reembed_wiki_page( return resp +@celery.task(**DURABLE_TASK) +@with_idempotency(task_name="convert_source_to_wiki") +def convert_source_to_wiki(self, source_id, user, idempotency_key=None): + from application.worker import convert_source_to_wiki_worker + + resp = convert_source_to_wiki_worker(self, source_id, user) + return resp + + # Beat-driven dispatch tasks default to ``acks_late=False``: a SIGKILL # of a beat tick is harmless to redeliver only if the dispatch itself is # idempotent. We keep these early-ACK so the broker doesn't replay a diff --git a/application/storage/db/source_config.py b/application/storage/db/source_config.py index 540073e7..845968af 100644 --- a/application/storage/db/source_config.py +++ b/application/storage/db/source_config.py @@ -117,6 +117,19 @@ class SourceConfig(BaseModel): chunking: ChunkingConfig = ChunkingConfig() retrieval: RetrievalConfig = RetrievalConfig() + def wiki_enabled(self) -> dict: + """Return a config dict flipped to wiki mode + browse-as-you-go exposure. + + Sets ``kind="wiki"`` and defaults ``retrieval.exposure`` to + ``agentic_tool`` (a wiki is navigated, not bulk-prefetched), preserving + any non-default exposure the source already carries. + """ + new_config = self.model_dump() + new_config["kind"] = "wiki" + if self.retrieval.exposure == "prefetch": + new_config["retrieval"]["exposure"] = "agentic_tool" + return new_config + @classmethod def parse(cls, raw: Optional[dict]) -> "SourceConfig": """Lenient read: return all-defaults for ``{}``/``None``. diff --git a/application/worker.py b/application/worker.py index 7e61bc9c..0a8bc98e 100755 --- a/application/worker.py +++ b/application/worker.py @@ -39,7 +39,11 @@ from application.storage.db.repositories.ingest_chunk_progress import ( IngestChunkProgressRepository, ) from application.storage.db.repositories.sources import SourcesRepository -from application.storage.db.repositories.wiki_pages import WikiPagesRepository +from application.storage.db.repositories.wiki_pages import ( + WikiPagesRepository, + _content_hash, + rebuild_wiki_directory_structure, +) from application.storage.db.session import db_readonly, db_session from application.storage.db.source_config import SourceConfig from application.storage.storage_creator import StorageCreator @@ -174,6 +178,25 @@ def _apply_display_names_to_structure(structure, file_name_map, prefix=""): return structure +def _download_source_files_to_dir(storage, source_file_path, temp_dir): + """Mirror a source's stored files into ``temp_dir``, preserving structure.""" + if not storage.is_directory(source_file_path): + return + for storage_file_path in storage.list_files(source_file_path): + if storage.is_directory(storage_file_path): + continue + rel_path = os.path.relpath(storage_file_path, source_file_path) + local_file_path = os.path.join(temp_dir, rel_path) + os.makedirs(os.path.dirname(local_file_path), exist_ok=True) + try: + file_data = storage.get_file(storage_file_path) + with open(local_file_path, "wb") as f: + f.write(file_data.read()) + except Exception as e: + logging.error(f"Error downloading file {storage_file_path}: {e}") + continue + + # Define a function to generate a random string of a given length. @@ -734,29 +757,7 @@ def reingest_source_worker(self, source_id, user): ) with tempfile.TemporaryDirectory() as temp_dir: - # Download all files from storage to temp directory, preserving directory structure - if storage.is_directory(source_file_path): - files_list = storage.list_files(source_file_path) - - for storage_file_path in files_list: - if storage.is_directory(storage_file_path): - continue - - rel_path = os.path.relpath(storage_file_path, source_file_path) - local_file_path = os.path.join(temp_dir, rel_path) - - os.makedirs(os.path.dirname(local_file_path), exist_ok=True) - - # Download file - try: - file_data = storage.get_file(storage_file_path) - with open(local_file_path, "wb") as f: - f.write(file_data.read()) - except Exception as e: - logging.error( - f"Error downloading file {storage_file_path}: {e}" - ) - continue + _download_source_files_to_dir(storage, source_file_path, temp_dir) reader = SimpleDirectoryReader( input_dir=temp_dir, @@ -2058,3 +2059,145 @@ def reembed_wiki_page_worker(self, source_id, path, content_hash, user): raise return {"status": "embedded", "added": added, "deleted": deleted} + + +def _wiki_page_path_from_rel(rel_path): + """Build a validated leading-slash wiki page path, or None if invalid. + + Runs the derived path through ``validate_tool_path`` so a stored filename + carrying traversal (``..``) never lands as a ``wiki_pages.path``. + """ + from application.agents.tools.path_utils import validate_tool_path + + raw = "/" + rel_path.replace(os.sep, "/").lstrip("/") + return validate_tool_path(raw) + + +def convert_source_to_wiki_worker(self, source_id, user): + """Convert an ingested source into a wiki by materializing files as pages. + + Re-parses the source's stored originals (the same storage file-load + + ``SimpleDirectoryReader`` machinery reingest uses), creating one wiki page + per file with extractable text. Non-text/binary/unsupported files are + skipped and reported. Embedding is enqueued per page (async); the source's + ``config.kind`` flips to ``wiki`` only after at least one page is + materialized — a source with no extractable pages keeps its original kind + and directory structure. + + Args: + self: Celery task instance. + source_id: ID of the source to convert. + user: Owner identifier used to load the source row and author pages. + + Returns: + A summary dict ``{"status": "converted", "pages_created", "skipped"}``, + ``{"status": "no_pages", ...}`` when nothing was extractable, or + ``{"status": "already_wiki"}`` when the source is already a wiki. + """ + from application.api.user.tasks import reembed_wiki_page + + source_id = str(source_id) + + with db_readonly() as conn: + source = SourcesRepository(conn).get_any(source_id, user) + if not source: + raise ValueError(f"Source {source_id} not found or access denied") + source_id = str(source["id"]) + + cfg = SourceConfig.parse(source.get("config")) + if cfg.kind == "wiki": + return {"status": "already_wiki", "pages_created": 0, "skipped": []} + + storage = StorageCreator.get_storage() + source_file_path = source.get("file_path", "") + file_name_map = _normalize_file_name_map(source.get("file_name_map")) + + created_pages: list[tuple[str, str]] = [] + skipped: list[dict] = [] + + stored_files: set[str] = set() + if storage.is_directory(source_file_path): + for stored in storage.list_files(source_file_path): + if storage.is_directory(stored): + continue + stored_files.add(os.path.relpath(stored, source_file_path)) + + with tempfile.TemporaryDirectory() as temp_dir: + _download_source_files_to_dir(storage, source_file_path, temp_dir) + + reader = SimpleDirectoryReader( + input_dir=temp_dir, + recursive=True, + required_exts=list(SUPPORTED_SOURCE_EXTENSIONS), + exclude_hidden=True, + errors="ignore", + file_metadata=metadata_from_filename, + ) + raw_docs = reader.load_data() + + texts_by_rel: dict[str, list[str]] = {} + for doc in raw_docs: + extra_info = getattr(doc, "extra_info", None) or {} + rel_path = extra_info.get("source") + if not rel_path: + continue + text = getattr(doc, "text", None) or "" + if not text.strip(): + continue + texts_by_rel.setdefault(rel_path, []).append(text) + + with db_session() as conn: + repo = WikiPagesRepository(conn) + for rel_path in sorted(texts_by_rel): + page_path = _wiki_page_path_from_rel(rel_path) + if page_path is None: + skipped.append({"file": rel_path, "reason": "invalid path"}) + continue + content = "\n\n".join(texts_by_rel[rel_path]) + title = _get_display_name(file_name_map, rel_path) or os.path.basename( + rel_path + ) + repo.upsert( + source_id, page_path, content, title=title, updated_by=user + ) + created_pages.append((page_path, _content_hash(content))) + + for rel_path in sorted(stored_files - set(texts_by_rel)): + ext = os.path.splitext(rel_path)[1].lower() + reason = ( + "could not read/parse" + if ext in SUPPORTED_SOURCE_EXTENSIONS + else "no extractable text" + ) + skipped.append({"file": rel_path, "reason": reason}) + + if created_pages: + with db_session() as conn: + rebuild_wiki_directory_structure(conn, source_id, user) + + if not created_pages: + return { + "status": "no_pages", + "pages_created": 0, + "skipped": skipped, + } + + for page_path, content_hash in created_pages: + reembed_wiki_page.delay( + source_id, + page_path, + content_hash, + user=user, + idempotency_key=f"reembed-wiki:{source_id}:{page_path}:{content_hash}", + ) + + with db_session() as conn: + SourcesRepository(conn).update( + source_id, user, {"config": cfg.wiki_enabled()} + ) + + return { + "status": "converted", + "pages_created": len(created_pages), + "skipped": skipped, + } diff --git a/tests/api/user/sources/test_routes.py b/tests/api/user/sources/test_routes.py index e3653c0f..739e1220 100644 --- a/tests/api/user/sources/test_routes.py +++ b/tests/api/user/sources/test_routes.py @@ -1029,3 +1029,108 @@ class TestSourceConfigResource: # owner-scoped read sees it. got = SourcesRepository(pg_conn).get_any(sid, owner) assert got["config"]["retrieval"]["chunks"] == 9 + + def test_kind_flip_to_wiki_rejected(self, app, pg_conn): + # Flipping kind to wiki must route through /wiki/convert, not config. + from application.api.user.sources.routes import SourceConfigResource + from application.storage.db.repositories.sources import SourcesRepository + + user = "u-cfg-wiki-flip" + src = _seed_source(pg_conn, user, name="cfg-wiki", type="file") + sid = str(src["id"]) + + with _patch_db(pg_conn), app.test_request_context( + f"/api/sources/{sid}/config", + method="PATCH", + json={"kind": "wiki"}, + ): + from flask import request + request.decoded_token = {"sub": user} + response = SourceConfigResource().patch(sid) + + assert response.status_code == 400 + # The kind must NOT have silently flipped. + from application.storage.db.source_config import SourceConfig + + got = SourcesRepository(pg_conn).get_any(sid, user) + assert SourceConfig.parse(got.get("config")).kind != "wiki" + + def test_other_edits_work_on_wiki_source(self, app, pg_conn): + # A wiki source can still edit retrieval (kind stays wiki, no reject). + from application.api.user.sources.routes import SourceConfigResource + from application.storage.db.repositories.sources import SourcesRepository + + user = "u-cfg-wiki-edit" + src = _seed_source( + pg_conn, user, name="wiki-cfg", type="wiki", + config={"kind": "wiki"}, + ) + sid = str(src["id"]) + + with _patch_db(pg_conn), app.test_request_context( + f"/api/sources/{sid}/config", + method="PATCH", + json={"kind": "wiki", "retrieval": {"chunks": 4}}, + ): + from flask import request + request.decoded_token = {"sub": user} + response = SourceConfigResource().patch(sid) + + assert response.status_code == 200 + got = SourcesRepository(pg_conn).get_any(sid, user) + assert got["config"]["retrieval"]["chunks"] == 4 + + def test_partial_edit_preserves_wiki_kind(self, app, pg_conn): + # A partial edit that OMITS kind must not demote a wiki to classic + # (SourceConfig.kind defaults to "classic" on a full-replace write). + from application.api.user.sources.routes import SourceConfigResource + from application.storage.db.repositories.sources import SourcesRepository + from application.storage.db.source_config import SourceConfig + + user = "u-cfg-wiki-partial" + src = _seed_source( + pg_conn, user, name="wiki-partial", type="wiki", + config={"kind": "wiki"}, + ) + sid = str(src["id"]) + + with _patch_db(pg_conn), app.test_request_context( + f"/api/sources/{sid}/config", + method="PATCH", + json={"retrieval": {"chunks": 6}}, + ): + from flask import request + request.decoded_token = {"sub": user} + response = SourceConfigResource().patch(sid) + + assert response.status_code == 200 + assert response.json["config"]["kind"] == "wiki" + got = SourcesRepository(pg_conn).get_any(sid, user) + assert SourceConfig.parse(got.get("config")).kind == "wiki" + assert got["config"]["retrieval"]["chunks"] == 6 + + def test_explicit_kind_demotion_from_wiki_rejected(self, app, pg_conn): + # Demoting wiki -> classic via config is rejected; use /wiki/convert. + from application.api.user.sources.routes import SourceConfigResource + from application.storage.db.repositories.sources import SourcesRepository + from application.storage.db.source_config import SourceConfig + + user = "u-cfg-wiki-demote" + src = _seed_source( + pg_conn, user, name="wiki-demote", type="wiki", + config={"kind": "wiki"}, + ) + sid = str(src["id"]) + + with _patch_db(pg_conn), app.test_request_context( + f"/api/sources/{sid}/config", + method="PATCH", + json={"kind": "classic"}, + ): + from flask import request + request.decoded_token = {"sub": user} + response = SourceConfigResource().patch(sid) + + assert response.status_code == 400 + got = SourcesRepository(pg_conn).get_any(sid, user) + assert SourceConfig.parse(got.get("config")).kind == "wiki" diff --git a/tests/api/user/sources/test_wiki_routes.py b/tests/api/user/sources/test_wiki_routes.py index 2233660c..89ce19d6 100644 --- a/tests/api/user/sources/test_wiki_routes.py +++ b/tests/api/user/sources/test_wiki_routes.py @@ -290,3 +290,276 @@ class TestWikiPage: request.decoded_token = {"sub": stranger} response = WikiPage().get(sid) assert response.status_code == 404 + + +class TestConvertSourceToWiki: + def test_returns_401_unauthenticated(self, app): + from application.api.user.sources.routes import ConvertSourceToWiki + + with app.test_request_context( + "/api/sources/x/wiki/convert", method="POST" + ): + from flask import request + request.decoded_token = None + response = ConvertSourceToWiki().post("x") + assert response.status_code == 401 + + def test_blank_source_enabled_inline_no_task(self, app, pg_conn): + from application.api.user.sources.routes import ConvertSourceToWiki + from application.storage.db.repositories.sources import SourcesRepository + from application.storage.db.source_config import SourceConfig + + user = "u-convert-blank" + src = SourcesRepository(pg_conn).create( + "blank", user_id=user, type="file", directory_structure={} + ) + sid = str(src["id"]) + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.convert_source_to_wiki.delay" + ) as mock_convert, app.test_request_context( + f"/api/sources/{sid}/wiki/convert", method="POST" + ): + from flask import request + request.decoded_token = {"sub": user} + response = ConvertSourceToWiki().post(sid) + + assert response.status_code == 200 + assert response.json["converted"] is False + assert response.json["enabled"] is True + mock_convert.assert_not_called() + got = SourcesRepository(pg_conn).get_any(sid, user) + cfg = SourceConfig.parse(got.get("config")) + assert cfg.kind == "wiki" + assert cfg.retrieval.exposure == "agentic_tool" + + def test_fileful_source_enqueues_task(self, app, pg_conn): + from application.api.user.sources.routes import ConvertSourceToWiki + from application.storage.db.repositories.sources import SourcesRepository + + user = "u-convert-files" + src = SourcesRepository(pg_conn).create( + "files", user_id=user, type="file", + directory_structure={"a.md": {"type": "text/markdown"}}, + ) + sid = str(src["id"]) + + fake_task = type("T", (), {"id": "task-xyz"})() + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.convert_source_to_wiki.delay", + return_value=fake_task, + ) as mock_convert, app.test_request_context( + f"/api/sources/{sid}/wiki/convert", method="POST" + ): + from flask import request + request.decoded_token = {"sub": user} + response = ConvertSourceToWiki().post(sid) + + assert response.status_code == 200 + assert response.json["task_id"] == "task-xyz" + mock_convert.assert_called_once() + assert mock_convert.call_args.kwargs["source_id"] == sid + assert mock_convert.call_args.kwargs["user"] == user + assert mock_convert.call_args.kwargs["idempotency_key"] == ( + f"convert-wiki:{sid}" + ) + + def test_in_progress_ingest_rejected_409(self, app, pg_conn): + from application.api.user.sources.routes import ConvertSourceToWiki + from application.storage.db.repositories.ingest_chunk_progress import ( + IngestChunkProgressRepository, + ) + from application.storage.db.repositories.sources import SourcesRepository + + user = "u-convert-ingesting" + src = SourcesRepository(pg_conn).create( + "ingesting", user_id=user, type="file", + directory_structure={"a.md": {"type": "text/markdown"}}, + ) + sid = str(src["id"]) + # An active embed in flight (embedded 0 of 5) → "processing". + IngestChunkProgressRepository(pg_conn).init_progress(sid, 5) + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.convert_source_to_wiki.delay" + ) as mock_convert, app.test_request_context( + f"/api/sources/{sid}/wiki/convert", method="POST" + ): + from flask import request + request.decoded_token = {"sub": user} + response = ConvertSourceToWiki().post(sid) + + assert response.status_code == 409 + mock_convert.assert_not_called() + + def test_viewer_rejected_403(self, app, pg_conn): + from application.api.user.sources.routes import ConvertSourceToWiki + from application.storage.db.repositories.sources import SourcesRepository + + owner = "alice-convert" + viewer = "bob-convert-viewer" + src = SourcesRepository(pg_conn).create( + "files", user_id=owner, type="file", + directory_structure={"a.md": {"type": "text/markdown"}}, + ) + sid = str(src["id"]) + _grant_team_access(pg_conn, owner, viewer, sid, "viewer") + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.convert_source_to_wiki.delay" + ) as mock_convert, app.test_request_context( + f"/api/sources/{sid}/wiki/convert", method="POST" + ): + from flask import request + request.decoded_token = {"sub": viewer} + response = ConvertSourceToWiki().post(sid) + + assert response.status_code == 403 + mock_convert.assert_not_called() + + +class TestWikiPageEdit: + def test_returns_401_unauthenticated(self, app): + from application.api.user.sources.routes import WikiPage + + with app.test_request_context( + "/api/sources/x/wiki/page", method="PUT", json={} + ): + from flask import request + request.decoded_token = None + response = WikiPage().put("x") + assert response.status_code == 401 + + def test_owner_writes_and_enqueues_reembed(self, app, pg_conn): + from application.api.user.sources.routes import WikiPage + from application.storage.db.repositories.sources import SourcesRepository + from application.storage.db.repositories.wiki_pages import WikiPagesRepository + + user = "u-edit-owner" + src = SourcesRepository(pg_conn).create( + "wiki", user_id=user, type="wiki", config={"kind": "wiki"} + ) + sid = str(src["id"]) + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.reembed_wiki_page.delay" + ) as mock_reembed, app.test_request_context( + f"/api/sources/{sid}/wiki/page", + method="PUT", + json={"path": "/notes.md", "content": "body"}, + ): + from flask import request + request.decoded_token = {"sub": user} + response = WikiPage().put(sid) + + assert response.status_code == 200 + assert response.json["page"]["path"] == "/notes.md" + assert response.json["page"]["version"] is not None + page = WikiPagesRepository(pg_conn).get_by_path(sid, "/notes.md") + assert page["content"] == "body" + mock_reembed.assert_called_once() + assert mock_reembed.call_args.args[1] == "/notes.md" + assert mock_reembed.call_args.kwargs["user"] == user + + def test_team_editor_reembeds_as_owner(self, app, pg_conn): + from application.api.user.sources.routes import WikiPage + from application.storage.db.repositories.sources import SourcesRepository + + owner = "alice-edit" + editor = "bob-edit-editor" + src = SourcesRepository(pg_conn).create( + "wiki", user_id=owner, type="wiki", config={"kind": "wiki"} + ) + sid = str(src["id"]) + _grant_team_access(pg_conn, owner, editor, sid, "editor") + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.reembed_wiki_page.delay" + ) as mock_reembed, app.test_request_context( + f"/api/sources/{sid}/wiki/page", + method="PUT", + json={"path": "/e.md", "content": "edited"}, + ): + from flask import request + request.decoded_token = {"sub": editor} + response = WikiPage().put(sid) + + assert response.status_code == 200 + # Re-embed runs AS the owner so the owner-scoped worker load resolves. + assert mock_reembed.call_args.kwargs["user"] == owner + + def test_stale_version_returns_409(self, app, pg_conn): + from application.api.user.sources.routes import WikiPage + from application.storage.db.repositories.sources import SourcesRepository + from application.storage.db.repositories.wiki_pages import WikiPagesRepository + + user = "u-edit-conflict" + src = SourcesRepository(pg_conn).create( + "wiki", user_id=user, type="wiki", config={"kind": "wiki"} + ) + sid = str(src["id"]) + WikiPagesRepository(pg_conn).upsert(sid, "/c.md", "v1", updated_by=user) + # Bump to version 2 so a stale expected_version=1 loses the race. + WikiPagesRepository(pg_conn).upsert(sid, "/c.md", "v2", updated_by=user) + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.reembed_wiki_page.delay" + ) as mock_reembed, app.test_request_context( + f"/api/sources/{sid}/wiki/page", + method="PUT", + json={"path": "/c.md", "content": "v3", "expected_version": 1}, + ): + from flask import request + request.decoded_token = {"sub": user} + response = WikiPage().put(sid) + + assert response.status_code == 409 + mock_reembed.assert_not_called() + + def test_traversal_path_returns_400(self, app, pg_conn): + from application.api.user.sources.routes import WikiPage + from application.storage.db.repositories.sources import SourcesRepository + + user = "u-edit-traversal" + src = SourcesRepository(pg_conn).create( + "wiki", user_id=user, type="wiki", config={"kind": "wiki"} + ) + sid = str(src["id"]) + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.reembed_wiki_page.delay" + ), app.test_request_context( + f"/api/sources/{sid}/wiki/page", + method="PUT", + json={"path": "/../secret.md", "content": "x"}, + ): + from flask import request + request.decoded_token = {"sub": user} + response = WikiPage().put(sid) + assert response.status_code == 400 + + def test_viewer_rejected_403(self, app, pg_conn): + from application.api.user.sources.routes import WikiPage + from application.storage.db.repositories.sources import SourcesRepository + + owner = "alice-edit-viewer" + viewer = "bob-edit-viewer" + src = SourcesRepository(pg_conn).create( + "wiki", user_id=owner, type="wiki", config={"kind": "wiki"} + ) + sid = str(src["id"]) + _grant_team_access(pg_conn, owner, viewer, sid, "viewer") + + with _patch_db(pg_conn), patch( + "application.api.user.sources.routes.reembed_wiki_page.delay" + ) as mock_reembed, app.test_request_context( + f"/api/sources/{sid}/wiki/page", + method="PUT", + json={"path": "/v.md", "content": "x"}, + ): + from flask import request + request.decoded_token = {"sub": viewer} + response = WikiPage().put(sid) + + assert response.status_code == 403 + mock_reembed.assert_not_called() diff --git a/tests/api/user/test_tasks.py b/tests/api/user/test_tasks.py index a27e5856..b38c065a 100644 --- a/tests/api/user/test_tasks.py +++ b/tests/api/user/test_tasks.py @@ -86,6 +86,20 @@ class TestReingestSourceTask: assert result == {"status": "ok"} +class TestConvertSourceToWikiTask: + @pytest.mark.unit + @patch("application.worker.convert_source_to_wiki_worker") + def test_calls_convert_worker(self, mock_worker): + from application.api.user.tasks import convert_source_to_wiki + + mock_worker.return_value = {"status": "converted"} + + result = convert_source_to_wiki("source123", "user1") + + mock_worker.assert_called_once_with(ANY, "source123", "user1") + assert result == {"status": "converted"} + + class TestScheduleSyncsTask: @pytest.mark.unit @patch("application.api.user.tasks.sync_worker") @@ -286,6 +300,8 @@ class TestDurableTaskRetryPolicy: "store_attachment", "process_agent_webhook", "ingest_connector_task", + "reembed_wiki_page", + "convert_source_to_wiki", ], ) def test_task_has_retry_config(self, task_name): diff --git a/tests/worker/test_convert_source_to_wiki.py b/tests/worker/test_convert_source_to_wiki.py new file mode 100644 index 00000000..8347b55b --- /dev/null +++ b/tests/worker/test_convert_source_to_wiki.py @@ -0,0 +1,305 @@ +"""Tests for ``application.worker.convert_source_to_wiki_worker``. + +The worker re-parses a source's stored originals into wiki pages and enqueues +a per-page re-embed. Storage, ``SimpleDirectoryReader``, and the re-embed task +are mocked so no real downloads or embeddings run; the ``sources`` / +``wiki_pages`` rows are real so the assertions read them back. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from application.parser.schema.base import Document +from application.storage.db.repositories.sources import SourcesRepository +from application.storage.db.repositories.wiki_pages import WikiPagesRepository + + +def _seed_source(pg_conn, config=None, file_path="inputs/alice/doc-set"): + src = SourcesRepository(pg_conn).create( + "doc-set", + user_id="alice", + type="local", + retriever="classic", + file_path=file_path, + config=config or {}, + directory_structure={"a.md": {"type": "text/markdown", "size_bytes": 1}}, + ) + return str(src["id"]) + + +def _patch_storage(monkeypatch, stored_files): + """Storage whose directory holds ``stored_files`` (rel paths).""" + from application import worker + + base = "inputs/alice/doc-set" + storage = MagicMock(name="storage") + storage.is_directory.side_effect = lambda p: p == base + storage.list_files.return_value = [f"{base}/{f}" for f in stored_files] + storage.get_file.return_value = MagicMock(read=lambda: b"x") + monkeypatch.setattr(worker.StorageCreator, "get_storage", lambda: storage) + return storage + + +def _patch_reader(monkeypatch, docs): + from application import worker + + reader = MagicMock(name="reader") + reader.load_data.return_value = docs + monkeypatch.setattr(worker, "SimpleDirectoryReader", lambda *a, **kw: reader) + return reader + + +@pytest.mark.unit +class TestConvertSourceToWikiWorker: + def test_one_page_per_text_file( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_storage(monkeypatch, ["guide.md", "notes/intro.md", "logo.png"]) + _patch_reader( + monkeypatch, + [ + Document(text="guide body", extra_info={"source": "guide.md"}), + Document( + text="intro body", extra_info={"source": "notes/intro.md"} + ), + ], + ) + delay = MagicMock(name="reembed_delay") + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", delay + ) + + result = worker.convert_source_to_wiki_worker( + task_self, source_id, "alice" + ) + + assert result["status"] == "converted" + assert result["pages_created"] == 2 + + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + by_path = {p["path"]: p for p in pages} + assert set(by_path) == {"/guide.md", "/notes/intro.md"} + assert by_path["/guide.md"]["content"] == "guide body" + assert by_path["/guide.md"]["title"] == "guide.md" + assert by_path["/notes/intro.md"]["title"] == "intro.md" + + def test_binaries_skipped_and_reported( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_storage(monkeypatch, ["guide.md", "logo.png", "scan.pdf"]) + _patch_reader( + monkeypatch, + [Document(text="guide body", extra_info={"source": "guide.md"})], + ) + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", MagicMock() + ) + + result = worker.convert_source_to_wiki_worker( + task_self, source_id, "alice" + ) + + assert result["pages_created"] == 1 + skipped_files = {s["file"] for s in result["skipped"]} + assert skipped_files == {"logo.png", "scan.pdf"} + for entry in result["skipped"]: + assert entry["reason"] + + def test_reembed_enqueued_per_page_with_per_page_key( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_storage(monkeypatch, ["a.md", "b.md"]) + _patch_reader( + monkeypatch, + [ + Document(text="aaa", extra_info={"source": "a.md"}), + Document(text="bbb", extra_info={"source": "b.md"}), + ], + ) + delay = MagicMock(name="reembed_delay") + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", delay + ) + + worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + assert delay.call_count == 2 + keys = {c.kwargs["idempotency_key"] for c in delay.call_args_list} + paths = {c.args[1] for c in delay.call_args_list} + assert paths == {"/a.md", "/b.md"} + for call in delay.call_args_list: + assert call.args[0] == source_id + assert call.kwargs["user"] == "alice" + assert call.kwargs["idempotency_key"].startswith( + f"reembed-wiki:{source_id}:" + ) + assert len(keys) == 2 + + def test_sets_kind_wiki_and_agentic_exposure( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + from application.storage.db.source_config import SourceConfig + + source_id = _seed_source(pg_conn) + _patch_storage(monkeypatch, ["a.md"]) + _patch_reader( + monkeypatch, + [Document(text="aaa", extra_info={"source": "a.md"})], + ) + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", MagicMock() + ) + + worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + refreshed = SourcesRepository(pg_conn).get_any(source_id, "alice") + cfg = SourceConfig.parse(refreshed.get("config")) + assert cfg.kind == "wiki" + assert cfg.retrieval.exposure == "agentic_tool" + + def test_preserves_non_default_exposure( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + from application.storage.db.source_config import SourceConfig + + source_id = _seed_source( + pg_conn, config={"retrieval": {"exposure": "agentic_tool"}} + ) + _patch_storage(monkeypatch, ["a.md"]) + _patch_reader( + monkeypatch, + [Document(text="aaa", extra_info={"source": "a.md"})], + ) + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", MagicMock() + ) + + worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + refreshed = SourcesRepository(pg_conn).get_any(source_id, "alice") + cfg = SourceConfig.parse(refreshed.get("config")) + assert cfg.kind == "wiki" + assert cfg.retrieval.exposure == "agentic_tool" + + def test_no_pages_leaves_kind_and_structure( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + from application.storage.db.source_config import SourceConfig + + original_structure = {"a.md": {"type": "text/markdown", "size_bytes": 1}} + source_id = _seed_source(pg_conn) + _patch_storage(monkeypatch, ["logo.png", "scan.pdf"]) + _patch_reader(monkeypatch, []) + rebuild = MagicMock(name="rebuild") + monkeypatch.setattr( + worker, "rebuild_wiki_directory_structure", rebuild + ) + delay = MagicMock(name="reembed_delay") + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", delay + ) + + result = worker.convert_source_to_wiki_worker( + task_self, source_id, "alice" + ) + + assert result["status"] == "no_pages" + assert result["pages_created"] == 0 + assert result["skipped"] + rebuild.assert_not_called() + delay.assert_not_called() + + refreshed = SourcesRepository(pg_conn).get_any(source_id, "alice") + assert SourceConfig.parse(refreshed.get("config")).kind != "wiki" + assert refreshed.get("directory_structure") == original_structure + + def test_traversal_filename_skipped( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_storage(monkeypatch, ["ok.md", "../evil.md"]) + _patch_reader( + monkeypatch, + [ + Document(text="good", extra_info={"source": "ok.md"}), + Document(text="bad", extra_info={"source": "../evil.md"}), + ], + ) + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", MagicMock() + ) + + result = worker.convert_source_to_wiki_worker( + task_self, source_id, "alice" + ) + + assert result["pages_created"] == 1 + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + assert {p["path"] for p in pages} == {"/ok.md"} + skipped = {(s["file"], s["reason"]) for s in result["skipped"]} + assert ("../evil.md", "invalid path") in skipped + + def test_parse_failure_reason_distinct_from_binary( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + # broken.md has a supported extension but produced no doc (parse/read + # failure); blob.bin is an unsupported/binary file. + _patch_storage(monkeypatch, ["guide.md", "broken.md", "blob.bin"]) + _patch_reader( + monkeypatch, + [Document(text="guide body", extra_info={"source": "guide.md"})], + ) + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", MagicMock() + ) + + result = worker.convert_source_to_wiki_worker( + task_self, source_id, "alice" + ) + + reasons = {s["file"]: s["reason"] for s in result["skipped"]} + assert reasons["broken.md"] == "could not read/parse" + assert reasons["blob.bin"] == "no extractable text" + + def test_already_wiki_returns_early( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn, config={"kind": "wiki"}) + reader = MagicMock(name="reader") + monkeypatch.setattr( + worker, "SimpleDirectoryReader", lambda *a, **kw: reader + ) + delay = MagicMock(name="reembed_delay") + monkeypatch.setattr( + "application.api.user.tasks.reembed_wiki_page.delay", delay + ) + + result = worker.convert_source_to_wiki_worker( + task_self, source_id, "alice" + ) + + assert result["status"] == "already_wiki" + reader.load_data.assert_not_called() + delay.assert_not_called()