diff --git a/.claude/skills/taos-development-skill/SKILL.md b/.claude/skills/taos-development-skill/SKILL.md index d2515b222..75f7d1dc2 100644 --- a/.claude/skills/taos-development-skill/SKILL.md +++ b/.claude/skills/taos-development-skill/SKILL.md @@ -240,6 +240,15 @@ A finding folded within hours merges the same day; a finding left while you open new PRs stalls the whole train behind it (the maintainer will not merge past an open finding, ever). +Folding means code AND a reply: answer every numbered item in the PR thread +with the commit that addresses it or a concrete rebuttal. Code pushed without +in-thread replies leaves the fold formally open and the verdict at HOLD. + +Bot review freshness is part of folding: check WHICH commit a bot actually +reviewed. A rate-limited or stale "SUCCESS" on an older head is not a pass - +after pushing fixes, re-trigger the review (for CodeRabbit: comment +`@coderabbitai review`). + ### Rebase cadence and the stale-base rule - dev moves fast. When your PR shows CONFLICTING, rebase onto current dev @@ -265,6 +274,64 @@ A close without a successor link reads as lost work and forces the maintainer into git forensics (this happened with #1927/#1924 - both were legitimate "landed via" closures that looked like data loss for hours). +### Asking another team's agent a question + +taOS depends on sibling services with their own maintainer agents (taOSmd, the +website). Their contracts are theirs to state, so ASK rather than inferring from +their source (pitfall 23). How to reach them depends on where you sit: + +- **On the A2A bus** (internal agents): post on the relevant channel, name the + agent, and expect a reply inside the hour. +- **Outside the bus** (external contributors): open an issue on + `jaylfc/taos-agent-commons`, the private invite-only coordination repo. Label + it `contract-question` and name the service. @taOS-dev sweeps it hourly and + relays to the owning agent on the bus, then carries the answer back. +- `jaylfc/taosmd` is public with issues enabled, so a taosmd contract question + can also go straight there. +- `jaylfc/taos-website` is private, so commons or the relay is the only route. + +If a question sits unanswered for more than about two hours, escalate by also +raising it on the PR. The relay is a person-shaped hop and can stall; silence +should never be mistaken for progress. + +This arrangement is TEMPORARY scaffolding. It retires when an external +contributor can hold a taOS identity and reach the bus directly, which is the +same capability as agent sharing. Do not build tooling that assumes it is +permanent. + +### Verify before you claim, and compile before you PR + +- An assertion with a tolerance (`abs(a - b) <= n`) does not test an invariant. + It cannot tell correct behaviour from total failure. Assert equality and + assert the resulting state (pitfall 20). +- Mocking internals or injecting unreproducible errors is fine. Mocking an + external service CONTRACT is where tests lie. For sibling services (taOSmd, + the website) the contract has a reachable owner on the A2A bus: ASK them + rather than inferring from their source. Every mock failure in the #2062 + cycle was a guess at something one message would have answered. External + contributors ask in the PR and the lead relays. For genuine third parties, + capture a real response and commit it rather than composing a fixture from + what your code expects. Then keep one feature detecting integration test per + contract, or mark the mock provisional in code with its source and date. A + follow-up issue does not count: it separates the caveat from the code. A + green test over a fictional contract certifies the bug (pitfall 23). +- Typecheck or run the thing before opening the PR. A frontend change that does + not compile wastes a full review round, and the executor now gates on + `tsc --noEmit` for exactly that reason (pitfall 21). +- When the base branch has moved under you, rebase and read what changed before + resolving conflicts. Taking the wrong side of a hunk silently reverts fixes + that were just merged (pitfall 22). + +### Scope honesty per slice + +- The PR body states exactly what it ships versus what its issue scopes. Any + deferred part gets an explicit deferral note AND a follow-up issue filed in + the same push; the parent issue never auto-closes on a partial slice. +- Before committing, `git status` must show only intended source changes - + runtime artifacts (anything under `data/`) never enter a commit, and a new + component that writes under `data/` gitignores its directory in the same PR + (pitfall 16). + ### One PR per slice - A fix and the test that proves it belong in ONE PR (pitfall 13 in diff --git a/.gitignore b/.gitignore index bb7dfe4d3..91b3770ac 100644 --- a/.gitignore +++ b/.gitignore @@ -30,6 +30,8 @@ data/videos/ data/secrets.db data/models/ data/workspace/ +data/library/ +data/collections/ data/*.log # Environment diff --git a/CHANGELOG.md b/CHANGELOG.md index 5159980b8..261f8827e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,42 @@ Versions follow semver beta: `1.0.0-beta.N`, bumped on each dev->master promotio ## [Unreleased] +## [1.0.0-beta.43] - 2026-07-21 + +### Added + +- **Library app P1**: LibraryStore, ingest pipeline, cheap-tier processors (file, + text, PDF, image) and the collections handoff to taOSmd. Ingested items are + copied into a per-item directory and registered as a taOSmd collection over the + live Collections API, with async index polling and typed link rows (#2062). +- Invite mint accepts an optional `ttl_secs` (60s to 24h) so a longer-lived + invite is a deliberate choice rather than a code change (#2072). + +### Fixed + +- **Invite dialog crashed the desktop.** The invite list endpoints returned + `scopes` as a JSON-encoded string while the UI typed it as an array, so the + dialog threw and tripped the SPA error boundary whenever any pending invite + existed. This also caused the post-mint refresh to unmount the dialog before + the URL and PIN were shown (#2066). +- **Expired invites could not be revoked.** `revoke` only matched `pending`, so an + invite that lazily flipped to `expired` returned 404 while still listed, leaving + dead rows against the pending cap. Revoke now covers expired, and terminal + states return 409 with the actual state (#2071). +- Default invite TTL raised from 15 minutes to 1 hour. Handing a URL and PIN to a + human who then configures an agent is not a 15 minute flow (#2072). + +### Changed + +- `docs/getting-started.md` documents the Hailo-10H AI HAT+2 and the Raspberry Pi + 5 M.2 slot conflict: the HAT occupies the only M.2 slot, so it cannot be used + alongside an NVMe boot drive (#2075). +- Contributor docs gained seven new defect classes drawn from real review + findings, covering runtime state in commits, mobile view registries, retrofit + migrations on shipped stores, scope honesty, tolerance assertions, shell + snippets in template literals, conflict resolution, and how an external + contributor reaches another team's agent (#2069, #2079). + ## [1.0.0-beta.42] - 2026-07-20 ### Added diff --git a/desktop/package.json b/desktop/package.json index 68ec685f8..8e213ebba 100644 --- a/desktop/package.json +++ b/desktop/package.json @@ -1,7 +1,7 @@ { "name": "tinyagentos-desktop", "private": true, - "version": "1.0.0-beta.42", + "version": "1.0.0-beta.43", "type": "module", "scripts": { "dev": "vite", diff --git a/docs/contributor-pitfalls.md b/docs/contributor-pitfalls.md index 54591933d..b65b265cc 100644 --- a/docs/contributor-pitfalls.md +++ b/docs/contributor-pitfalls.md @@ -12,6 +12,10 @@ session + CSRF, or the agent-token allowlist in `auth_middleware.py`) and a test asserting an unauthenticated or cross-principal caller gets 401/403. Example: PR #2036 added `GET /api/secrets/agent/{name}/github` with no guard, leaking installation IDs and repo names to any caller. +For project-scoped routes, copy the house guard verbatim: +`if not user.is_admin and user.user_id != p["user_id"]` followed by a masked +404 (see routes/projects.py). A bare 403 without the admin bypass both locks +out admins and leaks resource existence to other users (#2042). **2. Bind both directions on authenticated channels.** When a message or envelope arrives over an authenticated channel, verify BOTH @@ -37,6 +41,16 @@ and strips it. Tokens we only VERIFY are stored hashed; tokens we must PRESENT outbound are the only ones stored recoverable. Example: PR #2009 moved the GitHub App RSA key out of plaintext config.yaml. +**16. Never commit runtime state or key material.** +Anything a store or subsystem writes under `data/` at runtime is state, not +source. A PR that introduces a component writing under `data/` must add that +directory to `.gitignore` in the SAME PR, and `git status` must be checked for +runtime artifacts before every commit. Any private key that reaches a pushed +commit is burned: regenerate it, and keep it out of the target branch's history +(squash merge, or rewrite the branch). Example: `data/hub/identity.json` with +live signing/encryption keys entered one branch's history and was then +re-committed by a second PR (#2043 history, #2042). + ## Correctness **5. Never take element `[0]` of a collection that can hold more than one.** @@ -74,6 +88,12 @@ share flow), state it in the PR body and file a follow-up issue. Example: PR permissions. Wire the real value or do not add the parameter yet. Example: PR #2036 `handleSaveGrants`. +**17. A new view must be wired into every surface it has: desktop AND mobile.** +The desktop tab list and the mobile tab order are separate registries; updating +one and not the other ships a view that is unreachable on phones (#2042: +`TABS` updated, `mobileTabOrder` missed). Grep for every registry the sibling +views appear in and update all of them. + ## Store and schema **11. SCHEMA is the frozen v1; new columns and their indexes go in MIGRATIONS.** @@ -87,6 +107,18 @@ Running a data migration twice must be a no-op (existence checks, INSERT OR IGNORE, migrated-from markers) and there must be a test proving the second run changes nothing. Example done right: PR #2028 `test_migrate_idempotent`. +**18. Retrofitting a column onto an already-shipped store needs a guarded +ALTER, not just a MIGRATIONS entry.** +The migration runner baselines pre-existing databases at the latest version +WITHOUT executing the migrations (FOOTGUN #2 in `db_migrations.py`'s own +docstring), so a plain `MIGRATIONS = [(1, "ALTER TABLE ...")]` on a store that +already shipped is a silent no-op on every upgraded install: fresh DBs work, +upgraded DBs lose the feature at runtime. Use the guarded `_post_init` pattern +(PRAGMA table_info, ALTER only when the column is absent - see +`knowledge_store._migration_v1_add_user_id`) and add an upgrade test that +builds the PRE-change schema first. Fresh-DB tests are structurally blind to +this class (#2043: `peer_fingerprint`). + ## Process **13. A fix and its test belong in one PR.** @@ -110,3 +142,87 @@ Findings are gated on severity of content, not review state. Address each one (fix it or rebut it concretely in the thread); never merge past an open finding, and never let a "pass" verdict from a stale commit stand in for the current head. +A finding is closed only when it is ANSWERED IN-THREAD: reply to each numbered +item with the commit that addresses it or a concrete rebuttal. Pushing code +without replies leaves the finding open - the reviewer re-verifies blind and +the verdict stays HOLD (#2043 round two). + +**19. State scope deltas against the issue explicitly.** +If a PR ships less than its issue scopes - a subfeature dropped, owner-only +routes where the issue says peer-serving, a "live" view that fetches once - +say so in the PR body and file the follow-up issue in the same push. Never let +a partial slice close the parent issue. Silent shortfalls read as done, get +caught in review anyway, and cost a full extra round (#2042: community chat +and peer access absent with no deferral note). + +**20. A tolerance is not a test of an invariant.** +An assertion like `assert abs(after - before) <= 2` cannot distinguish correct +behaviour from catastrophic failure. In PR #2062 that exact assertion ran green +while reprocess destroyed the user's original uploaded file: the observed values +were `before=2, after=0`, which the tolerance accepted, and the same tolerance +would equally have accepted a doubling to 4. If the property is "the count does +not change", assert equality and assert the resulting status, so the test fails +loudly on both loss and duplication. Reserve tolerances for genuinely +approximate quantities such as timings, and even then bound them tightly. + +**21. Shell snippets inside template literals must escape `${`.** +A bash or PowerShell snippet stored in a JavaScript template literal collides +with the language's own interpolation: `${VAR:-default}` is parsed as JS, not +shell. In PR #2077 this produced 225 TypeScript syntax errors from a single +cause, in one of four otherwise-clean files, and read like incoherent output +rather than one mechanical mistake. Escape as `\${`, or keep snippets in plain +non-template strings or separate asset files. The class generalises to any +language sharing `${...}` with the shell, and it is invisible to anything that +does not actually compile the file, which is why the frontend typecheck gate +exists before a PR is opened. + +**22. Conflict resolution is a decision, not a mechanical act.** +Taking the wrong side of a hunk silently reverts fixes that were just made. When +a base branch has moved substantially under a long-lived branch, read what +changed underneath before resolving: the four defects fixed in Library P1 +(nested response envelope, the guard preventing reprocess from deleting the +source upload, the compare-and-swap status transition, exact artifact-count +assertions) are all reintroducible by a plausible-looking resolution. Rebase +rather than merging the base in, so the diff stays reviewable and each conflict +is seen individually. + +**23. Mocks of EXTERNAL service contracts need a real source, not a guess.** +Mocking our own internals or injecting errors you cannot produce on demand (a +500, a timeout, an ImportError) is fine and unavoidable. The dangerous class is +narrow: a mock of a service we do not control, whose fixture was hand-written +from what the calling code expects. That fixture encodes a BELIEF about someone +else's API, and when the belief is wrong the test proves nothing while going +green. This happened three times in one PR cycle (#2062): an invented `dbPath` +request shape, an un-nested poll response, and a tolerance assertion over both. + +First, split the class by whether the contract has a reachable OWNER. + +**Sibling services (taOSmd, taOS website): ASK. Do not guess.** These are not +third parties. Their maintainer is on the A2A bus, and asking costs one message. +Every failure in the #2062 cycle came from inferring a contract that was free to +obtain: the builder guessed a request shape, and the reviewer verified against +source rather than asking the owner. When we finally asked, we got the envelope +documented, a wrong stats key list corrected by its own author, and a +`/version` capabilities endpoint built to make the contract machine-checkable. +Reverse-engineering a sibling's API from its source is a smell, not diligence: +source tells you what it does today, the owner tells you what it guarantees. +Contributors without bus access (external collaborators) ask in the PR, and the +lead relays. + +**Genuinely third-party (GitHub, OpenRouter, Reddit): capture, do not compose.** +There is no one to ask, so call the real service once and commit its response as +the fixture. A recorded response cannot encode a wrong belief. + +For both, then: + +1. **Keep one real integration test per external contract.** Have it feature + detect and skip when the service is unreachable, so CI stays green offline + but drift surfaces the moment anyone runs it against a live instance. For + taosmd, `GET /version` returns a capabilities list for exactly this. +2. **If that is impossible, mark the mock provisional IN CODE** with the + contract source and the date it was verified. A follow-up issue is not + sufficient: it detaches the caveat from the code and, in a tracker with + hundreds of open items, functions as indefinite deferral. + +Reviewer's job: ask where the fixture came from. A green test over a fictional +contract is worse than no test, because it certifies the bug. diff --git a/docs/getting-started.md b/docs/getting-started.md index de9da39a7..fcb0b8f00 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -22,7 +22,7 @@ Any of the following will work. More RAM means bigger, more capable models. |--------|-----|-------| | **Orange Pi 5 Plus** (recommended) | 16 GB | RK3588 chip with 6 TOPS NPU for fast inference | | **Orange Pi 5** | 8–16 GB | Same NPU, slightly fewer I/O ports | -| **Raspberry Pi 5** | 8 GB minimum | CPU-only inference unless you add an accelerator HAT | +| **Raspberry Pi 5** | 8 GB minimum | CPU-only inference unless you add an accelerator HAT; with the **AI HAT+2 (Hailo-10H, 40 TOPS)** you get NPU-accelerated LLM inference | | **Any x86/x64 PC or laptop** | 4 GB+ | Budget PC, old laptop, NUC, etc. GPU optional | | **NVIDIA GPU system** | 4 GB+ VRAM | GTX 1050 Ti and up; CUDA acceleration | | **AMD GPU system** | 8 GB+ VRAM | RX 6600 and up; ROCm acceleration | @@ -31,6 +31,8 @@ The platform itself uses roughly 345 MB of RAM when idle, so it runs comfortably **Not sure which to buy?** The Orange Pi 5 Plus with 16 GB is the recommended starting point. It has a built-in NPU (neural processing unit) that runs models significantly faster than the CPU alone, and 16 GB gives you room to run multiple agents at once. +> **AI HAT+2 (Hailo-10H) and the Pi 5 M.2 slot:** The AI HAT+2 NPU accelerator occupies the Raspberry Pi 5's only M.2 slot, the same slot normally used for an NVMe boot drive. You cannot use the HAT and an NVMe SSD at the same time. If you want fast storage alongside Hailo LLM acceleration, use a USB SSD instead. + ### Software - **OS:** Armbian or Debian-based Linux (Ubuntu works too). The installer handles everything else. diff --git a/pyproject.toml b/pyproject.toml index 2760144f4..09a2ede4c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "tinyagentos" -version = "1.0.0-beta.42" +version = "1.0.0-beta.43" description = "Self-hosted AI agent memory system for low-power hardware" license = { file = "LICENSE" } # Upper-capped at <3.14 because litellm (the proxy extra, the agent/model proxy diff --git a/tests/projects/test_invite_store.py b/tests/projects/test_invite_store.py index 509404a19..0d5c3a77f 100644 --- a/tests/projects/test_invite_store.py +++ b/tests/projects/test_invite_store.py @@ -315,6 +315,44 @@ async def test_revoke_nonexistent_returns_false(store): assert await store.revoke("000000") is False +@pytest.mark.asyncio +async def test_revoke_expired_invite(store): + """An expired invite can still be revoked (cleanup from the pending list).""" + result = await store.mint( + project_id="prj-1", + scopes=[], + approval_mode="auto", + check_interval_secs=1800, + created_by="u", + ) + iid = result["record"]["invite_id"] + await store._db.execute( + "UPDATE project_invites SET expires_ts = 1 WHERE invite_id = ?", (iid,) + ) + await store._db.commit() + row = await store.get(iid) + assert row["status"] == "expired" + ok = await store.revoke(iid) + assert ok is True + row = await store.get(iid) + assert row["status"] == "revoked" + + +@pytest.mark.asyncio +async def test_revoke_redeemed_invite_returns_false(store): + result = await store.mint( + project_id="prj-1", + scopes=[], + approval_mode="auto", + check_interval_secs=1800, + created_by="u", + ) + iid = result["record"]["invite_id"] + pin = result["pin"] + await store.redeem(iid, pin) + assert await store.revoke(iid) is False + + # --------------------------------------------------------------------------- # redeem # --------------------------------------------------------------------------- diff --git a/tests/test_library.py b/tests/test_library.py new file mode 100644 index 000000000..37100dab6 --- /dev/null +++ b/tests/test_library.py @@ -0,0 +1,735 @@ +"""Tests for the Library app — store, pipeline, routes, and collections handoff.""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient +from tinyagentos.library_pipeline import ( + FileProcessor, + ImageProcessor, + PdfProcessor, + TextProcessor, + detect_kind, + run_pipeline, +) +from tinyagentos.library_store import LibraryStore +from tinyagentos.library_collections import handoff_to_collections + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest_asyncio.fixture +async def lib_store(): + """Create a LibraryStore backed by a temporary SQLite database.""" + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = Path(f.name) + + store = LibraryStore(db_path) + await store.init() + + yield store + + await store.close() + try: + db_path.unlink() + except OSError: + pass + + +@pytest.fixture +def storage_dir(): + """Create a temporary directory for file artifacts.""" + with tempfile.TemporaryDirectory() as d: + yield Path(d) + + +# --------------------------------------------------------------------------- +# Kind detection +# --------------------------------------------------------------------------- + + +class TestKindDetection: + def test_detect_youtube_url(self): + assert detect_kind(source_url="https://www.youtube.com/watch?v=abc123") == "url:youtube" + assert detect_kind(source_url="https://youtube.com/watch?v=abc123") == "url:youtube" + assert detect_kind(source_url="https://youtu.be/abc123") == "url:youtube" + assert detect_kind(source_url="https://m.youtube.com/watch?v=abc123") == "url:youtube" + + def test_detect_web_url(self): + assert detect_kind(source_url="https://example.com") == "url:web" + assert detect_kind(source_url="http://blog.example.com/post") == "url:web" + + def test_detect_by_mime(self): + assert detect_kind(content_type="text/plain") == "text" + assert detect_kind(content_type="application/pdf") == "pdf" + assert detect_kind(content_type="image/png") == "image" + assert detect_kind(content_type="image/jpeg") == "image" + assert detect_kind(content_type="application/zip") == "archive" + + def test_detect_by_filename(self): + assert detect_kind(file_path="doc.txt") == "text" + assert detect_kind(file_path="report.pdf") == "pdf" + assert detect_kind(file_path="photo.jpg") == "image" + assert detect_kind(file_path="icon.png") == "image" + + def test_detect_fallback(self): + assert detect_kind(file_path="unknown.xyz") == "file" + assert detect_kind() == "file" + + +# --------------------------------------------------------------------------- +# LibraryStore +# --------------------------------------------------------------------------- + + +class TestLibraryStore: + @pytest.mark.asyncio + async def test_create_and_get_item(self, lib_store): + item_id = await lib_store.create_item( + kind="text", + title="test.txt", + source_url="", + storage_path="/tmp/test.txt", + size_bytes=42, + ) + assert item_id + + item = await lib_store.get_item(item_id) + assert item is not None + assert item["kind"] == "text" + assert item["title"] == "test.txt" + assert item["status"] == "pending" + assert item["bytes"] == 42 + + @pytest.mark.asyncio + async def test_get_nonexistent_item(self, lib_store): + item = await lib_store.get_item("nonexistent") + assert item is None + + @pytest.mark.asyncio + async def test_list_items(self, lib_store): + id1 = await lib_store.create_item(kind="text", title="a.txt") + id2 = await lib_store.create_item(kind="pdf", title="b.pdf") + id3 = await lib_store.create_item(kind="image", title="c.png") + + items = await lib_store.list_items() + assert len(items) == 3 + + text_items = await lib_store.list_items(kind="text") + assert len(text_items) == 1 + assert text_items[0]["title"] == "a.txt" + + assert len(await lib_store.list_items(status="pending")) == 3 + assert len(await lib_store.list_items(status="ready")) == 0 + + @pytest.mark.asyncio + async def test_update_item(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="old") + await lib_store.update_item(item_id, title="new", status="ready") + + item = await lib_store.get_item(item_id) + assert item["title"] == "new" + assert item["status"] == "ready" + + @pytest.mark.asyncio + async def test_update_item_meta(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + await lib_store.update_item(item_id, meta_json={"preview": "hello"}) + + item = await lib_store.get_item(item_id) + meta = json.loads(item["meta_json"]) + assert meta["preview"] == "hello" + + @pytest.mark.asyncio + async def test_update_invalid_status(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + with pytest.raises(ValueError): + await lib_store.update_item_status(item_id, "invalid_status") + + @pytest.mark.asyncio + async def test_delete_item(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + item = await lib_store.get_item(item_id) + assert item is not None + + await lib_store.delete_item(item_id) + item = await lib_store.get_item(item_id) + assert item is None + + @pytest.mark.asyncio + async def test_artifacts(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + art_id = await lib_store.add_artifact(item_id, kind="text", path="/tmp/test.txt") + + artifacts = await lib_store.get_artifacts(item_id) + assert len(artifacts) == 1 + assert artifacts[0]["kind"] == "text" + + await lib_store.delete_artifact(art_id) + assert len(await lib_store.get_artifacts(item_id)) == 0 + + @pytest.mark.asyncio + async def test_cascade_delete_artifacts(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + await lib_store.add_artifact(item_id, kind="text", path="/tmp/a.txt") + await lib_store.add_artifact(item_id, kind="thumbnail", path="/tmp/thumb.jpg") + + await lib_store.delete_item(item_id) + assert len(await lib_store.get_artifacts(item_id)) == 0 + + @pytest.mark.asyncio + async def test_jobs(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + job_id = await lib_store.create_job(item_id, "ingest") + + job = await lib_store.get_job(job_id) + assert job is not None + assert job["stage"] == "ingest" + assert job["state"] == "queued" + + await lib_store.update_job(job_id, state="done") + job = await lib_store.get_job(job_id) + assert job["state"] == "done" + + +# --------------------------------------------------------------------------- +# Pipeline processors +# --------------------------------------------------------------------------- + + +class TestFileProcessor: + @pytest.mark.asyncio + async def test_process_existing_file(self, lib_store, storage_dir): + file_path = storage_dir / "test.txt" + file_path.write_text("hello world") + + item_id = await lib_store.create_item( + kind="file", title="test.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = FileProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 1 + assert artifacts[0]["kind"] == "metadata" + + @pytest.mark.asyncio + async def test_process_missing_file(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="file", title="missing.txt", storage_path="/nonexistent/file.txt" + ) + item = await lib_store.get_item(item_id) + + proc = FileProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 0 + + +class TestTextProcessor: + @pytest.mark.asyncio + async def test_extract_text(self, lib_store, storage_dir): + file_path = storage_dir / "notes.txt" + file_path.write_text("line one\nline two\nline three") + + item_id = await lib_store.create_item( + kind="text", title="notes.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 1 + assert artifacts[0]["kind"] == "text" + assert artifacts[0]["meta"]["line_count"] == 3 + assert artifacts[0]["meta"]["char_count"] == 28 + + text_path = Path(artifacts[0]["path"]) + assert text_path.exists() + assert "line one" in text_path.read_text() + + @pytest.mark.asyncio + async def test_text_auto_title(self, lib_store, storage_dir): + file_path = storage_dir / "notes.txt" + file_path.write_text("My Title\nmore content here") + + item_id = await lib_store.create_item( + kind="text", title="", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + await proc.process(item) + + updated = await lib_store.get_item(item_id) + assert updated["title"] == "My Title" + + @pytest.mark.asyncio + async def test_text_preview(self, lib_store, storage_dir): + file_path = storage_dir / "long.txt" + file_path.write_text("A" * 500) + + item_id = await lib_store.create_item( + kind="text", title="long.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + await proc.process(item) + + updated = await lib_store.get_item(item_id) + meta = json.loads(updated["meta_json"]) + assert "preview" in meta + assert len(meta["preview"]) == 200 + + +class TestPdfProcessor: + @pytest.mark.asyncio + async def test_process_pdf(self, lib_store, storage_dir): + file_path = storage_dir / "test.pdf" + _create_minimal_pdf(file_path) + + item_id = await lib_store.create_item( + kind="pdf", title="test.pdf", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = PdfProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + + meta_artifacts = [a for a in artifacts if a["kind"] == "metadata"] + assert len(meta_artifacts) >= 1 + assert meta_artifacts[0]["meta"]["page_count"] >= 0 + + @pytest.mark.asyncio + async def test_process_missing_pdf(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="pdf", title="missing.pdf", storage_path="/nonexistent/file.pdf" + ) + item = await lib_store.get_item(item_id) + + proc = PdfProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 0 + + +class TestImageProcessor: + @pytest.mark.asyncio + async def test_process_image(self, lib_store, storage_dir): + file_path = storage_dir / "test.png" + _create_test_image(file_path) + + item_id = await lib_store.create_item( + kind="image", title="test.png", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = ImageProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + + kinds = {a["kind"] for a in artifacts} + assert "metadata" in kinds + assert "thumbnail" in kinds + + thumb_art = [a for a in artifacts if a["kind"] == "thumbnail"][0] + assert Path(thumb_art["path"]).exists() + + @pytest.mark.asyncio + async def test_process_missing_image(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="image", title="missing.png", storage_path="/nonexistent/file.png" + ) + item = await lib_store.get_item(item_id) + + proc = ImageProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 0 + + +# --------------------------------------------------------------------------- +# run_pipeline +# --------------------------------------------------------------------------- + + +class TestRunPipeline: + @pytest.mark.asyncio + async def test_pipeline_text(self, lib_store, storage_dir): + file_path = storage_dir / "notes.txt" + file_path.write_text("sample content") + + item_id = await lib_store.create_item( + kind="text", title="notes.txt", storage_path=str(file_path) + ) + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + artifacts = await lib_store.get_artifacts(item_id) + artifact_kinds = {a["kind"] for a in artifacts} + assert "metadata" in artifact_kinds + assert "text" in artifact_kinds + + @pytest.mark.asyncio + async def test_pipeline_file(self, lib_store, storage_dir): + file_path = storage_dir / "unknown.xyz" + file_path.write_text("raw data") + + item_id = await lib_store.create_item( + kind="file", title="unknown.xyz", storage_path=str(file_path) + ) + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + @pytest.mark.asyncio + async def test_pipeline_error_status(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="file", title="missing.txt", storage_path="/missing/file.txt" + ) + await run_pipeline(lib_store, item_id, storage_dir) + item = await lib_store.get_item(item_id) + assert item["status"] == "error" + + +# --------------------------------------------------------------------------- +# Collections handoff +# --------------------------------------------------------------------------- + + +class TestCollectionsHandoff: + @pytest.mark.asyncio + async def test_handoff_files_copied(self, lib_store, storage_dir): + """Text artifacts are copied to collections dir even without qmd; + returns 0 when qmd is unreachable (no silent ImportError success).""" + file_path = storage_dir / "notes.txt" + file_path.write_text("content for collections") + + item_id = await lib_store.create_item( + kind="text", title="notes.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + await proc.process(item) + + collections_dir = storage_dir / "collections" + count = await handoff_to_collections(lib_store, item_id, collections_dir) + + # Files copied but no qmd → 0 indexed + assert count == 0 + + # Verify files were still copied to collections dir + item_dir = collections_dir / item_id + assert item_dir.exists() + text_files = list(item_dir.glob("*.txt")) + assert len(text_files) >= 1 + assert "content for collections" in text_files[0].read_text() + + @pytest.mark.asyncio + async def test_handoff_with_qmd(self, lib_store, storage_dir): + """Handoff indexes into taosmd when the API is reachable.""" + from unittest.mock import AsyncMock, patch + + file_path = storage_dir / "notes.txt" + file_path.write_text("hello from library") + + item_id = await lib_store.create_item( + kind="text", title="notes.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + await proc.process(item) + + collections_dir = storage_dir / "collections" + taosmd_url = "http://localhost:17900" + taosmd_token = "test-admin-token" + + # Mock httpx.AsyncClient to simulate a working taosmd + from unittest.mock import AsyncMock, MagicMock + + mock_client = MagicMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + + # POST /collections → nested response shape (taosmd 0.4.0) + mock_create_resp = MagicMock() + mock_create_resp.status_code = 200 + mock_create_resp.json.return_value = {"collection": {"id": "coll-123"}} + + # POST /collections/{id}/index → 202 (no body) + mock_index_resp = MagicMock() + mock_index_resp.status_code = 202 + + # POST /collections/{id}/link → 200 + mock_link_resp = MagicMock() + mock_link_resp.status_code = 200 + + # GET /collections/{id} → nested shape (taosmd 0.4.0), status=ready with stats + mock_poll_resp = MagicMock() + mock_poll_resp.status_code = 200 + mock_poll_resp.json.return_value = { + "collection": { + "status": "ready", + "stats": { + "files_indexed": 1, + "files_total": 1, + "chunks_ingested": 3, + "chunks_skipped": 0, + }, + }, + } + + mock_client.post = AsyncMock( + side_effect=[mock_create_resp, mock_index_resp, mock_link_resp] + ) + mock_client.get = AsyncMock(return_value=mock_poll_resp) + + with patch("httpx.AsyncClient", return_value=mock_client): + count = await handoff_to_collections( + lib_store, item_id, collections_dir, + taosmd_url=taosmd_url, + taosmd_admin_token=taosmd_token, + ) + + assert count == 1 # One file indexed per stats + + # Verify request sequence — create → index → link + assert mock_client.post.call_count == 3 + post_calls = [c.args[0] for c in mock_client.post.call_args_list] + assert post_calls[0] == f"{taosmd_url}/collections" + assert post_calls[1] == f"{taosmd_url}/collections/coll-123/index" + assert post_calls[2] == f"{taosmd_url}/collections/coll-123/link" + + # Verify POST /collections body contains required fields + create_kwargs = mock_client.post.call_args_list[0].kwargs + create_body = create_kwargs.get("json", {}) + assert create_body["name"] == f"library-{item_id[:12]}" + assert create_body["kind"] == "mixed" + assert create_body["source_path"] == str(collections_dir / item_id) + + # Verify poll GET called + mock_client.get.assert_called_once_with( + f"{taosmd_url}/collections/coll-123", + headers={"Authorization": f"Bearer {taosmd_token}"}, + ) + + +# --------------------------------------------------------------------------- +# API routes +# --------------------------------------------------------------------------- + + +class TestLibraryRoutes: + @pytest.mark.asyncio + async def test_library_page_gone(self, client): + """The server-rendered /library page was dropped (fold 6) — + the Library UI is the React desktop app.""" + resp = await client.get("/library") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_ingest_no_input(self, client): + resp = await client.post("/api/library/ingest") + assert resp.status_code == 400 + data = resp.json() + assert "error" in data + + @pytest.mark.asyncio + async def test_ingest_url(self, client): + resp = await client.post("/api/library/ingest", data={"url": "https://example.com/page"}) + assert resp.status_code == 202 + data = resp.json() + assert "item_id" in data + assert data["status"] == "pending" + + @pytest.mark.asyncio + async def test_ingest_file(self, client, tmp_path): + test_file = tmp_path / "hello.txt" + test_file.write_text("hello world") + + with open(test_file, "rb") as f: + resp = await client.post( + "/api/library/ingest", + files={"file": ("hello.txt", f, "text/plain")}, + ) + assert resp.status_code == 202 + data = resp.json() + assert "item_id" in data + + @pytest.mark.asyncio + async def test_list_items(self, client): + resp = await client.get("/api/library/items") + assert resp.status_code == 200 + data = resp.json() + assert "items" in data + assert "count" in data + + @pytest.mark.asyncio + async def test_get_nonexistent_item(self, client): + resp = await client.get("/api/library/items/nonexistent") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_delete_nonexistent_item(self, client): + resp = await client.delete("/api/library/items/nonexistent") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_reprocess_nonexistent_item(self, client): + resp = await client.post("/api/library/items/nonexistent/reprocess") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_ingest_and_get(self, client): + resp = await client.post("/api/library/ingest", data={"url": "https://example.com"}) + assert resp.status_code == 202 + item_id = resp.json()["item_id"] + + resp = await client.get(f"/api/library/items/{item_id}") + assert resp.status_code == 200 + data = resp.json() + assert data["item"]["id"] == item_id + assert "artifacts" in data + + @pytest.mark.asyncio + async def test_filter_by_kind(self, client): + await client.post("/api/library/ingest", data={"url": "https://example.com"}) + await client.post("/api/library/ingest", data={"url": "https://youtube.com/watch?v=abc"}) + + resp = await client.get("/api/library/items", params={"kind": "url:youtube"}) + data = resp.json() + for item in data["items"]: + assert item["kind"] == "url:youtube" + + # -- Auth: all endpoints require a session (CSRF is bypassed in tests, but + # unauthenticated requests still hit the global auth middleware). Test + # that the 6 new endpoints return 401 when no session cookie/header is + # present. + + @pytest.mark.asyncio + async def test_unauth_library_endpoints(self, app): + """All 6 library endpoints return 401 without authentication.""" + from httpx import ASGITransport, AsyncClient + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as unauth_client: + endpoints: list[tuple[str, str]] = [ + ("post", "/api/library/ingest"), + ("get", "/api/library/items"), + ("get", "/api/library/items/abc123"), + ("delete", "/api/library/items/abc123"), + ("post", "/api/library/items/abc123/reprocess"), + ] + for method, url in endpoints: + if method == "post": + resp = await unauth_client.post(url, data={"url": "https://example.com"}) + elif method == "delete": + resp = await unauth_client.delete(url) + else: + resp = await unauth_client.get(url) + assert resp.status_code == 401, f"{method.upper()} {url} returned {resp.status_code}, expected 401" + + # -- 413: oversized upload (tested at the HTTP level in integration; + # ASGI transport does not propagate Content-Length to file.size) + + @pytest.mark.asyncio + async def test_reprocess_idempotent(self, client, tmp_path): + """Reprocessing an item deletes old artifacts first — no duplicates.""" + test_file = tmp_path / "notes.txt" + test_file.write_text("test content") + + with open(test_file, "rb") as f: + resp = await client.post( + "/api/library/ingest", + files={"file": ("notes.txt", f, "text/plain")}, + ) + assert resp.status_code == 202 + item_id = resp.json()["item_id"] + + # Wait for pipeline to finish (background task) + import asyncio + await asyncio.sleep(0.5) + + # Check initial artifact count + resp = await client.get(f"/api/library/items/{item_id}") + data = resp.json() + initial_artifact_count = len(data["artifacts"]) + assert initial_artifact_count > 0, "Pipeline should produce artifacts" + + # First reprocess + resp = await client.post(f"/api/library/items/{item_id}/reprocess") + assert resp.status_code == 202 + await asyncio.sleep(0.5) + + # After first reprocess — exact same artifact count, not doubled. + resp = await client.get(f"/api/library/items/{item_id}") + data = resp.json() + after_first = len(data["artifacts"]) + assert after_first == initial_artifact_count, ( + f"Reprocess should not duplicate artifacts: " + f"initial={initial_artifact_count} after_first={after_first}" + ) + assert data["item"].get("status") == "ready", ( + f"Item status should be ready after reprocess, got {data['item'].get('status')}" + ) + + # Second reprocess — should still work, no duplicates + resp = await client.post(f"/api/library/items/{item_id}/reprocess") + assert resp.status_code == 202 + + @pytest.mark.asyncio + async def test_reprocess_while_processing_returns_409(self, client, app): + """Reprocessing an item that is already pending/processing returns 409.""" + # Ingest normally and wait for pipeline to finish + import asyncio + + resp = await client.post( + "/api/library/ingest", + data={"url": "https://example.com"}, + ) + assert resp.status_code == 202 + item_id = resp.json()["item_id"] + await asyncio.sleep(0.5) + + # Force status to "processing" to simulate an in-flight pipeline + store = app.state.library_store + await store.update_item_status(item_id, "processing") + + # Reprocess should be rejected + resp = await client.post(f"/api/library/items/{item_id}/reprocess") + assert resp.status_code == 409 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _create_minimal_pdf(path: Path): + """Create a minimal valid PDF file for testing.""" + pdf_content = ( + b"%PDF-1.4\n" + b"1 0 obj<>endobj\n" + b"2 0 obj<>endobj\n" + b"3 0 obj<>endobj\n" + b"xref\n0 4\n0000000000 65535 f \n0000000009 00000 n \n0000000058 00000 n \n0000000115 00000 n \n" + b"trailer<>\n" + b"startxref\n190\n%%EOF\n" + ) + path.write_bytes(pdf_content) + + +def _create_test_image(path: Path): + """Create a simple test image using PIL.""" + from PIL import Image + img = Image.new("RGB", (100, 50), color="blue") + img.save(path) diff --git a/tests/test_routes_project_invites.py b/tests/test_routes_project_invites.py index 60b67d280..5043baa52 100644 --- a/tests/test_routes_project_invites.py +++ b/tests/test_routes_project_invites.py @@ -139,6 +139,37 @@ async def test_revoke_nonexistent_returns_404(client, app): assert resp.status_code == 404, resp.text +@pytest.mark.asyncio +async def test_revoke_expired_returns_204(client, app): + pid = await _create_project(client) + mint_resp = await client.post( + f"/api/projects/{pid}/invites", + json={"scopes": [], "approval_mode": "auto"}, + ) + iid = mint_resp.json()["invite_id"] + store = app.state.project_invites + await store._db.execute( + "UPDATE project_invites SET expires_ts = 1 WHERE invite_id = ?", (iid,) + ) + await store._db.commit() + resp = await client.delete(f"/api/projects/{pid}/invites/{iid}") + assert resp.status_code == 204, resp.text + + +@pytest.mark.asyncio +async def test_revoke_already_revoked_returns_409(client, app): + pid = await _create_project(client) + mint_resp = await client.post( + f"/api/projects/{pid}/invites", + json={"scopes": [], "approval_mode": "auto"}, + ) + iid = mint_resp.json()["invite_id"] + first = await client.delete(f"/api/projects/{pid}/invites/{iid}") + assert first.status_code == 204, first.text + second = await client.delete(f"/api/projects/{pid}/invites/{iid}") + assert second.status_code == 409, second.text + + # --------------------------------------------------------------------------- # Redeem slice (S2): auto + manual approval, handle derivation, bundle, errors # --------------------------------------------------------------------------- @@ -585,3 +616,71 @@ async def test_guide_markdown_contains_required_instructions(client, app, monkey # Timed-check instruction mentions the interval value. assert "1800" in guide + + +@pytest.mark.asyncio +async def test_mint_default_ttl_is_one_hour(client, app): + pid = await _create_project(client) + before = time.time() + resp = await client.post( + f"/api/projects/{pid}/invites", + json={"scopes": [], "approval_mode": "auto"}, + ) + assert resp.status_code == 200, resp.text + ttl = resp.json()["expires_ts"] - before + assert 3500 < ttl <= 3610 + + +@pytest.mark.asyncio +async def test_mint_honours_ttl_secs(client, app): + pid = await _create_project(client) + before = time.time() + resp = await client.post( + f"/api/projects/{pid}/invites", + json={"scopes": [], "approval_mode": "auto", "ttl_secs": 86400}, + ) + assert resp.status_code == 200, resp.text + ttl = resp.json()["expires_ts"] - before + assert 86300 < ttl <= 86410 + + +@pytest.mark.asyncio +async def test_mint_rejects_ttl_over_cap(client, app): + pid = await _create_project(client) + resp = await client.post( + f"/api/projects/{pid}/invites", + json={"scopes": [], "approval_mode": "auto", "ttl_secs": 999999}, + ) + assert resp.status_code == 422, resp.text + + +@pytest.mark.asyncio +async def test_revoke_redeemed_returns_409(client, app): + pid = await _create_project(client) + mint_resp = await client.post( + f"/api/projects/{pid}/invites", + json={"scopes": [], "approval_mode": "auto"}, + ) + body = mint_resp.json() + store = app.state.project_invites + await store.redeem(body["invite_id"], body["pin"]) + resp = await client.delete(f"/api/projects/{pid}/invites/{body['invite_id']}") + assert resp.status_code == 409, resp.text + + +@pytest.mark.asyncio +async def test_revoke_claimed_returns_409_with_mid_redeem_message(client, app): + pid = await _create_project(client) + mint_resp = await client.post( + f"/api/projects/{pid}/invites", + json={"scopes": [], "approval_mode": "auto"}, + ) + iid = mint_resp.json()["invite_id"] + store = app.state.project_invites + await store._db.execute( + "UPDATE project_invites SET status = 'claimed' WHERE invite_id = ?", (iid,) + ) + await store._db.commit() + resp = await client.delete(f"/api/projects/{pid}/invites/{iid}") + assert resp.status_code == 409, resp.text + assert "mid-redeem" in resp.json()["error"] diff --git a/tinyagentos/__init__.py b/tinyagentos/__init__.py index fb06d1aea..a715928fc 100644 --- a/tinyagentos/__init__.py +++ b/tinyagentos/__init__.py @@ -1 +1 @@ -__version__ = "1.0.0-beta.42" +__version__ = "1.0.0-beta.43" diff --git a/tinyagentos/library_collections.py b/tinyagentos/library_collections.py new file mode 100644 index 000000000..c4cddb845 --- /dev/null +++ b/tinyagentos/library_collections.py @@ -0,0 +1,362 @@ +"""Library collections handoff — index text artifacts into taosmd collections. + +After the ingest pipeline produces text artifacts (extracted text, transcripts, +descriptions), this module hands them off to taosmd collections via the +taosmd HTTP API so agents can query the content through collection grants. + +The design doc (docs/design/library-app.md) says: + "Collections handoff: write text artifacts into a per-target folder under an + allowed root, then taosmd collections index; link to project; grants stay + EXPLICIT" + +Flow (Phase 1): + 1. Write text artifacts under collections_dir/{item_id}/ + 2. Create a collection via POST /collections (taosmd HTTP API) + 3. Trigger async index via POST /collections/{id}/index + 4. Poll GET /collections/{id} until status=ready|error + 5. Link collection to the library item via POST /collections/{id}/link +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import stat +from pathlib import Path + +logger = logging.getLogger(__name__) + +# Text artifact kinds that should be indexed into collections +_TEXT_ARTIFACT_KINDS = frozenset({"text", "transcript", "description", "ocr"}) + +# Maximum polls while waiting for async index to complete. +_MAX_INDEX_POLLS = 30 +# Seconds between poll attempts. +_POLL_INTERVAL = 2 + + +async def handoff_to_collections( + store, + item_id: str, + collections_dir: Path, + taosmd_url: str | None = None, + taosmd_admin_token: str | None = None, + project_id: str | None = None, +) -> int: + """Hand off all text artifacts for an item to taosmd collections. + + Writes text artifacts under ``collections_dir/{item_id}/``, then calls + the taosmd HTTP API to create a collection and index the text content. + + Returns the number of files indexed (``files_indexed`` from taosmd stats) + after a successful async index. Returns 0 when taosmd is unavailable + (no collection created). + + Parameters + ---------- + store: + LibraryStore instance. + item_id: + Library item id. + collections_dir: + Allowed root for collection files (e.g. ``/opt/taos/data/collections/``). + taosmd_url: + Base URL of the running taosmd instance (e.g. ``http://localhost:7900``). + When omitted or None, file-copy still happens but no collection is + created or indexed — the caller must have already created the collection + separately (production paths always supply this; test paths may omit it). + taosmd_admin_token: + Admin bearer token for taosmd API auth, fetched from SecretsStore as + ``taosmd-admin-token``. Required when *taosmd_url* is set. + project_id: + Optional project to link the collection to (Phase 2+). + """ + artifacts = await store.get_artifacts(item_id) + if not artifacts: + return 0 + + text_artifacts = [ + a for a in artifacts if a["kind"] in _TEXT_ARTIFACT_KINDS + ] + if not text_artifacts: + return 0 + + item = await store.get_item(item_id) + if not item: + return 0 + + # Write text artifacts to a per-item folder under the collections root + item_dir = collections_dir / item_id + item_dir.mkdir(parents=True, exist_ok=True) + # Set group-readable perms on the directory + try: + os.chmod(item_dir, stat.S_ISGID | stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP) + except OSError: + pass + + # Copy each text artifact into the collection source path. + # taosmd discovers files by scanning the source_path directory. + indexed_paths: list[str] = [] + for art in text_artifacts: + art_path = art.get("path", "") + if not art_path: + continue + + src = Path(art_path) + if not src.exists(): + continue + + dst = item_dir / src.name + try: + raw_bytes = src.read_bytes() + dst.write_bytes(raw_bytes) + # Group-readable file perms (0o640 — setgid meaningless on files) + try: + os.chmod(dst, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP) + except OSError: + pass + except OSError: + logger.warning("Failed to copy artifact %s → %s", src, dst, + exc_info=True) + continue + indexed_paths.append(str(dst)) + + if not indexed_paths: + return 0 + + # Index into taosmd collections via the taosmd HTTP API + if not taosmd_url: + logger.debug("taosmd URL not provided — collection indexing skipped") + return 0 + + # Parse item metadata once before the httpx try block so exception + # handlers (below) can merge into it without re-reading a stale item dict. + item_meta = json.loads(item.get("meta_json", "{}")) + + async def _set_retryable() -> None: + """Persist collection_retryable so a future retry can re-attempt. + + NOTE: collection_retryable currently has no consumer — it is scaffolding + for a future retry path. Clearing on success prevents stale markers + but no code path reads this flag yet. + """ + try: + item_meta["collection_retryable"] = True + await store.update_item(item_id, meta_json=item_meta) + except Exception: + pass + + try: + import httpx + + # Normalise base URL + base_url = taosmd_url.rstrip("/") + + # Build auth headers when a token is available + auth_headers: dict[str, str] = {} + if taosmd_admin_token: + auth_headers["Authorization"] = f"Bearer {taosmd_admin_token}" + + async with httpx.AsyncClient(timeout=30) as http_client: + # 1. Get or create a collection for this library item. + # Look up a previously-created collection id from item metadata + # so reprocess is idempotent (no duplicate collections). + collection_name = f"library-{item_id[:12]}" + title = item.get("title", "Untitled") or "Untitled" + + collection_id = "" + existing_coll_id = item_meta.get("collection_id", "") + + if existing_coll_id: + # Verify the collection still exists. + # 404 (genuinely gone) → fall through to create. + # Transient failure (5xx, timeout, connection error) → mark + # retryable and bail — must not create a duplicate. + try: + check_resp = await http_client.get( + f"{base_url}/collections/{existing_coll_id}", + headers=auth_headers, + ) + if check_resp.status_code < 400: + collection_id = existing_coll_id + logger.debug( + "Reusing existing collection %s for item %s", + collection_id, item_id, + ) + elif check_resp.status_code == 404: + logger.debug( + "Existing collection %s not found (404) — " + "will create replacement", existing_coll_id, + ) + else: + logger.warning( + "taosmd GET /collections/%s returned %d — " + "transient failure, marking retryable", + existing_coll_id, check_resp.status_code, + ) + await _set_retryable() + return 0 + except Exception: + logger.warning( + "taosmd GET /collections/%s failed — " + "transient failure, marking retryable", + existing_coll_id, exc_info=True, + ) + await _set_retryable() + return 0 + + if not collection_id: + source_path = str(item_dir) + create_resp = await http_client.post( + f"{base_url}/collections", + json={ + "name": collection_name, + "kind": "mixed", + "source_path": source_path, + "metadata": { + "title": title, + "kind": item.get("kind", ""), + "source_url": item.get("source_url", ""), + "library_item_id": item_id, + }, + }, + headers=auth_headers, + ) + if create_resp.status_code >= 400: + logger.warning( + "taosmd POST /collections returned %d for item %s: %s", + create_resp.status_code, item_id, create_resp.text[:200], + ) + await _set_retryable() + return 0 + # taosmd 0.4.0 nests the id under "collection" + collection_id = create_resp.json().get("collection", {}).get("id", "") + + if not collection_id: + logger.warning( + "taosmd POST /collections returned no collection.id for item %s", + item_id, + ) + await _set_retryable() + return 0 + + # Persist collection_id in item metadata for idempotent reprocess + item_meta["collection_id"] = collection_id + await store.update_item(item_id, meta_json=item_meta) + + # 2. Trigger async index — no body, returns 202. + try: + index_resp = await http_client.post( + f"{base_url}/collections/{collection_id}/index", + headers=auth_headers, + ) + if index_resp.status_code != 202: + logger.warning( + "taosmd POST /collections/%s/index returned %d (expected 202)", + collection_id, index_resp.status_code, + ) + await _set_retryable() + return 0 + except Exception: + logger.warning( + "taosmd POST /collections/%s/index failed", + collection_id, exc_info=True, + ) + await _set_retryable() + return 0 + + # 3. Poll until indexing completes. + indexed = 0 + for _poll_attempt in range(_MAX_INDEX_POLLS): + try: + poll_resp = await http_client.get( + f"{base_url}/collections/{collection_id}", + headers=auth_headers, + ) + if poll_resp.status_code >= 400: + logger.warning( + "taosmd GET /collections/%s returned %d during poll", + collection_id, poll_resp.status_code, + ) + await _set_retryable() + break + poll_data = poll_resp.json().get("collection", {}) + status = poll_data.get("status", "") + if status == "ready": + stats = poll_data.get("stats", {}) + indexed = stats.get("files_indexed", 0) + logger.info( + "Collections handoff for item %s: " + "files_indexed=%d files_total=%d " + "chunks_ingested=%d chunks_skipped=%d " + "collection=%s", + item_id, + stats.get("files_indexed", 0), + stats.get("files_total", 0), + stats.get("chunks_ingested", 0), + stats.get("chunks_skipped", 0), + collection_id, + ) + break + elif status == "error": + logger.warning( + "taosmd collection %s entered error state for item %s", + collection_id, item_id, + ) + await _set_retryable() + break + # Still indexing — wait and retry + await asyncio.sleep(_POLL_INTERVAL) + except Exception: + logger.warning( + "taosmd poll GET /collections/%s failed", + collection_id, exc_info=True, + ) + await _set_retryable() + break + else: + logger.warning( + "taosmd collection %s did not reach ready state within %d polls", + collection_id, _MAX_INDEX_POLLS, + ) + await _set_retryable() + + # 4. Link the collection to the library item. + try: + await http_client.post( + f"{base_url}/collections/{collection_id}/link", + json={"type": "taos", "id": item_id}, + headers=auth_headers, + ) + logger.debug( + "Linked collection %s to library item %s", + collection_id, item_id, + ) + except Exception: + logger.warning( + "taosmd POST /collections/%s/link failed for item %s", + collection_id, item_id, exc_info=True, + ) + + # Clear retryable on success; the handoff completed. + if indexed > 0: + item_meta.pop("collection_retryable", None) + try: + await store.update_item(item_id, meta_json=item_meta) + except Exception: + pass + + return indexed + + except ImportError: + logger.warning("httpx not available — collection indexing skipped") + await _set_retryable() + except Exception: + logger.exception( + "taosmd collections API unreachable for item %s", item_id, + ) + await _set_retryable() + + return 0 diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py new file mode 100644 index 000000000..e2ef7e185 --- /dev/null +++ b/tinyagentos/library_pipeline.py @@ -0,0 +1,428 @@ +"""Library ingest pipeline — processors for cheap-tier file/text/pdf/image ingestion. + +Processors are registered per detected kind and run asynchronously after ingest. +Each processor produces artifacts (e.g. metadata, extracted text, thumbnails) +that are stored on the item and optionally handed off to taosmd collections. +""" + +from __future__ import annotations + +import json +import logging +import mimetypes +import os +import time +from pathlib import Path + +from tinyagentos.library_store import LibraryStore + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Kind detection +# --------------------------------------------------------------------------- + +_MIME_KIND_MAP: dict[str, str] = { + "text/plain": "text", + "text/markdown": "text", + "text/csv": "text", + "text/html": "text", + "application/json": "text", + "application/xml": "text", + "text/xml": "text", + "application/pdf": "pdf", + "image/png": "image", + "image/jpeg": "image", + "image/gif": "image", + "image/webp": "image", + "image/svg+xml": "image", + "application/zip": "archive", + "application/gzip": "archive", + "application/x-tar": "archive", +} + + +def detect_kind(source_url: str = "", content_type: str = "", + file_path: str = "") -> str: + """Detect the library item kind from URL, MIME, or file path.""" + # URL-based detection + if source_url: + lower = source_url.lower() + if any(lower.startswith(p) for p in ("https://www.youtube.com/", + "https://youtube.com/", + "https://youtu.be/", + "https://m.youtube.com/")): + return "url:youtube" + if any(lower.startswith(p) for p in ("https://", "http://")): + return "url:web" + + # MIME-based detection + if content_type: + ct = content_type.split(";")[0].strip().lower() + if ct in _MIME_KIND_MAP: + return _MIME_KIND_MAP[ct] + + # File extension fallback + if file_path: + ext = Path(file_path).suffix.lower() + ext_map = { + ".txt": "text", ".md": "text", ".csv": "text", + ".json": "text", ".xml": "text", ".html": "text", + ".pdf": "pdf", + ".png": "image", ".jpg": "image", ".jpeg": "image", + ".gif": "image", ".webp": "image", ".svg": "image", + ".zip": "archive", ".gz": "archive", ".tar": "archive", + } + if ext in ext_map: + return ext_map[ext] + + return "file" + + +# --------------------------------------------------------------------------- +# Processor registry +# --------------------------------------------------------------------------- + +class Processor: + """Base processor. Subclasses handle one kind of library item.""" + + def __init__(self, store: LibraryStore, storage_dir: Path): + self.store = store + self.storage_dir = storage_dir + + async def process(self, item: dict) -> list[dict]: + """Run processing on an item, return list of artifact dicts produced.""" + raise NotImplementedError + + +class FileProcessor(Processor): + """Generic file processor — records basic metadata only.""" + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + source_url = item.get("source_url", "") + if storage_path: + p = Path(storage_path) + if p.exists(): + stat = p.stat() + file_meta = { + "size_bytes": stat.st_size, + "mtime": stat.st_mtime, + "source_url": source_url, + "processed_at": time.time(), + "processor": "FileProcessor/v1", + } + # Try mimetype detection + mime_type, _ = mimetypes.guess_type(p.name) + if mime_type: + file_meta["mime_type"] = mime_type + + # path="" because storage_path is the user's original uploaded + # file — the reprocess unlink loop must never delete it. + await self.store.add_artifact( + item_id, kind="metadata", path="", meta=file_meta + ) + artifacts.append({"kind": "metadata", "path": "", "meta": file_meta}) + + # Update item bytes + await self.store.update_item(item_id, bytes=stat.st_size) + + return artifacts + + +class TextProcessor(Processor): + """Text file processor — extracts content as text artifact.""" + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + if not storage_path: + return artifacts + + p = Path(storage_path) + if not p.exists(): + logger.warning("Text processor: file not found %s", storage_path) + return artifacts + + try: + text = p.read_text(encoding="utf-8", errors="replace") + except Exception: + logger.warning("Text processor: could not read %s", storage_path, + exc_info=True) + return artifacts + + # Write extracted text as an artifact + text_dir = self.storage_dir / "text" + text_dir.mkdir(parents=True, exist_ok=True) + text_path = text_dir / f"{item_id}.txt" + text_path.write_text(text, encoding="utf-8") + + text_meta = { + "char_count": len(text), + "line_count": text.count("\n") + 1, + "source_url": item.get("source_url", ""), + "processed_at": time.time(), + "processor": "TextProcessor/v1", + } + await self.store.add_artifact( + item_id, kind="text", path=str(text_path), meta=text_meta + ) + artifacts.append({"kind": "text", "path": str(text_path), "meta": text_meta}) + + # Store a preview (first 200 chars) + preview = text[:200] + meta = json.loads(item.get("meta_json", "{}")) + meta["preview"] = preview + await self.store.update_item(item_id, meta_json=meta) + + # Auto-title from content if no title + if not item.get("title"): + title = text.strip().split("\n", 1)[0][:100] + if title: + await self.store.update_item(item_id, title=title) + + return artifacts + + +class PdfProcessor(Processor): + """PDF processor — extracts page count and OCR-ready text (when available).""" + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + if not storage_path: + return artifacts + + p = Path(storage_path) + if not p.exists(): + return artifacts + + pdf_meta = {"page_count": 0, "has_text": False} + + # Try extracting text with PyPDF2 / pypdf if available + try: + from pypdf import PdfReader + reader = PdfReader(str(p)) + pdf_meta["page_count"] = len(reader.pages) + + # Extract text from all pages + pages_text: list[str] = [] + for page in reader.pages: + page_text = page.extract_text() + if page_text: + pages_text.append(page_text) + + if pages_text: + text_content = "\n\n".join(pages_text) + pdf_meta["has_text"] = True + pdf_meta["char_count"] = len(text_content) + + text_dir = self.storage_dir / "text" + text_dir.mkdir(parents=True, exist_ok=True) + text_path = text_dir / f"{item_id}_pdf.txt" + text_path.write_text(text_content, encoding="utf-8") + + await self.store.add_artifact( + item_id, kind="text", path=str(text_path), + meta={"char_count": len(text_content), "pages": len(reader.pages)}, + ) + artifacts.append({ + "kind": "text", "path": str(text_path), + "meta": {"char_count": len(text_content), "pages": len(reader.pages)}, + }) + + # Update item with preview + preview = text_content[:200] + meta = json.loads(item.get("meta_json", "{}")) + meta["preview"] = preview + await self.store.update_item(item_id, meta_json=meta) + except ImportError: + logger.debug("pypdf not installed — PDF text extraction skipped") + except Exception: + logger.warning("PDF text extraction failed for %s", storage_path, + exc_info=True) + + await self.store.add_artifact( + item_id, kind="metadata", path="", meta=pdf_meta + ) + artifacts.append({"kind": "metadata", "path": "", "meta": pdf_meta}) + + return artifacts + + +class ImageProcessor(Processor): + """Image processor — records dimensions, creates thumbnail. + + Thumbnail generation requires Pillow (PIL), which is always available in + the taOS dev dependencies. + """ + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + if not storage_path: + return artifacts + + p = Path(storage_path) + if not p.exists(): + return artifacts + + img_meta: dict = {"width": 0, "height": 0, "format": ""} + + try: + from PIL import Image + with Image.open(p) as img: + img_meta["width"] = img.width + img_meta["height"] = img.height + img_meta["format"] = img.format or "" + + # Create thumbnail (max 320px on longest side) + thumb_dir = self.storage_dir / "thumbs" + thumb_dir.mkdir(parents=True, exist_ok=True) + thumb_path = thumb_dir / f"{item_id}_thumb.jpg" + + img.thumbnail((320, 320)) + # Convert to RGB if needed (e.g. RGBA/PNG → JPEG) + if img.mode in ("RGBA", "P"): + img = img.convert("RGB") + img.save(thumb_path, "JPEG", quality=75) + + img_meta["thumbnail"] = str(thumb_path) + await self.store.add_artifact( + item_id, kind="thumbnail", path=str(thumb_path), + meta={"width": img.width, "height": img.height}, + ) + artifacts.append({ + "kind": "thumbnail", "path": str(thumb_path), + "meta": {"width": img.width, "height": img.height}, + }) + except ImportError: + logger.debug("PIL not available — image processing skipped") + except Exception: + logger.warning("Image processing failed for %s", storage_path, + exc_info=True) + + await self.store.add_artifact( + item_id, kind="metadata", path="", meta=img_meta + ) + artifacts.append({"kind": "metadata", "path": "", "meta": img_meta}) + + return artifacts + + +# --------------------------------------------------------------------------- +# Processor registry +# --------------------------------------------------------------------------- + +_PROCESSORS: dict[str, type[Processor]] = { + "file": FileProcessor, + "text": TextProcessor, + "pdf": PdfProcessor, + "image": ImageProcessor, +} + + +def get_processor(kind: str, store: LibraryStore, + storage_dir: Path) -> Processor: + """Return a processor for the given kind, falling back to FileProcessor.""" + cls = _PROCESSORS.get(kind, FileProcessor) + return cls(store, storage_dir) + + +# --------------------------------------------------------------------------- +# Pipeline runner +# --------------------------------------------------------------------------- + + +async def run_pipeline( + store: LibraryStore, + item_id: str, + storage_dir: Path, +) -> None: + """Run the ingest pipeline for one item. + + Steps: + 1. Mark item as 'processing' + 2. Determine processor from item kind + 3. Run file + kind-specific processors + 4. Collect artifacts + 5. Mark item as 'ready' (or 'error') + """ + item = await store.get_item(item_id) + if not item: + return + + kind = item["kind"] + + # URL-only items (no storage_path) are stored as references — the pipeline + # records a reference metadata artifact but does not fetch remote content + # (future: WebFetcherProcessor, #2078). The item gets a "reference" artifact so it + # is not silently empty. + if not item.get("storage_path") and item.get("source_url"): + logger.info( + "Library pipeline: URL-only item %s (%s) — stored as reference, not fetched", + item_id, kind, + ) + ref_meta = { + "source_url": item["source_url"], + "kind": kind, + "note": "Reference-only item — content not fetched (TODO: WebFetcherProcessor)", + } + await store.add_artifact( + item_id, kind="reference", path="", meta=ref_meta, + ) + + # If item has a storage_path that points to a missing file, fail early + # (dropped/moved/corrupt source must not silently look successful). + storage_path = item.get("storage_path", "") + source_url = item.get("source_url", "") + if storage_path and not source_url: + sp = Path(storage_path) + if not sp.exists(): + logger.warning( + "Library pipeline: source file missing for item %s: %s", + item_id, storage_path, + ) + await store.update_item_status(item_id, "error") + await store.update_item( + item_id, + meta_json={ + **json.loads(item.get("meta_json", "{}")), + "error": f"Source file not found: {storage_path}", + }, + ) + return + + try: + await store.update_item_status(item_id, "processing") + + # Stage 1: basic file metadata (always) + file_proc = FileProcessor(store, storage_dir) + await file_proc.process(item) + + # Stage 2: kind-specific processor + proc = get_processor(kind, store, storage_dir) + if not isinstance(proc, FileProcessor): + await proc.process(item) + + await store.update_item_status(item_id, "ready") + except Exception: + logger.exception("Library pipeline failed for item %s (kind=%s)", + item_id, kind) + await store.update_item_status(item_id, "error") + await store.update_item( + item_id, + meta_json={ + **json.loads(item.get("meta_json", "{}")), + "error": f"Pipeline failed for kind={kind}", + }, + ) diff --git a/tinyagentos/library_store.py b/tinyagentos/library_store.py new file mode 100644 index 000000000..6d7b1d11c --- /dev/null +++ b/tinyagentos/library_store.py @@ -0,0 +1,275 @@ +"""Persistent store for Library items, artifacts, and processing jobs. + +The Library is the universal ingestion surface for taOS — files, URLs, media +dropped into the Library get processed and their text artifacts indexed into +taosmd collections for agent access. + +Schema mirrors the design doc (docs/design/library-app.md): + - items: one row per ingested thing (file, URL, paste) + - artifacts: derived outputs (metadata, transcript, thumbnail, OCR text) + - jobs: async processing stages with retry support +""" + +from __future__ import annotations + +import json +import time +import uuid +from pathlib import Path + +import aiosqlite + +from tinyagentos.base_store import BaseStore + +LIBRARY_SCHEMA = """ +CREATE TABLE IF NOT EXISTS library_items ( + id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + source_url TEXT NOT NULL DEFAULT '', + title TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'pending', + storage_path TEXT NOT NULL DEFAULT '', + bytes INTEGER NOT NULL DEFAULT 0, + meta_json TEXT NOT NULL DEFAULT '{}', + created_at REAL NOT NULL, + updated_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_li_kind ON library_items(kind); +CREATE INDEX IF NOT EXISTS idx_li_status ON library_items(status); +CREATE INDEX IF NOT EXISTS idx_li_created ON library_items(created_at DESC); + +CREATE TABLE IF NOT EXISTS library_artifacts ( + id TEXT PRIMARY KEY, + item_id TEXT NOT NULL REFERENCES library_items(id) ON DELETE CASCADE, + kind TEXT NOT NULL, + path TEXT NOT NULL DEFAULT '', + meta_json TEXT NOT NULL DEFAULT '{}', + created_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_la_item ON library_artifacts(item_id); + +CREATE TABLE IF NOT EXISTS library_jobs ( + id TEXT PRIMARY KEY, + item_id TEXT NOT NULL REFERENCES library_items(id) ON DELETE CASCADE, + stage TEXT NOT NULL, + state TEXT NOT NULL DEFAULT 'queued', + error TEXT NOT NULL DEFAULT '', + created_at REAL NOT NULL, + updated_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_lj_item ON library_jobs(item_id); +CREATE INDEX IF NOT EXISTS idx_lj_state ON library_jobs(state); +""" + +_VALID_STATUSES = frozenset({"pending", "processing", "ready", "error"}) + + +class LibraryStore(BaseStore): + SCHEMA = LIBRARY_SCHEMA + + async def _post_init(self) -> None: + """Enable foreign key enforcement for cascade deletes.""" + await self._db.execute("PRAGMA foreign_keys = ON") + await self._db.commit() + + # -- items ------------------------------------------------------------ + + async def create_item( + self, + kind: str, + source_url: str = "", + title: str = "", + storage_path: str = "", + size_bytes: int = 0, + meta: dict | None = None, + ) -> str: + """Create a new library item. Returns the item id.""" + item_id = uuid.uuid4().hex + now = time.time() + await self._db.execute( + """INSERT INTO library_items + (id, kind, source_url, title, status, storage_path, bytes, + meta_json, created_at, updated_at) + VALUES (?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?)""", + ( + item_id, + kind, + source_url, + title, + storage_path, + size_bytes, + json.dumps(meta or {}), + now, + now, + ), + ) + await self._db.commit() + return item_id + + async def get_item(self, item_id: str) -> dict | None: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_items WHERE id = ?", (item_id,) + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + + async def list_items( + self, kind: str | None = None, status: str | None = None, + limit: int = 50, offset: int = 0, + ) -> list[dict]: + self._db.row_factory = aiosqlite.Row + where: list[str] = [] + params: list = [] + if kind: + where.append("kind = ?") + params.append(kind) + if status: + where.append("status = ?") + params.append(status) + + sql = "SELECT * FROM library_items" + if where: + sql += " WHERE " + " AND ".join(where) + sql += " ORDER BY created_at DESC LIMIT ? OFFSET ?" + params.extend([limit, offset]) + + async with self._db.execute(sql, params) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + async def update_item(self, item_id: str, **kwargs) -> None: + allowed = {"kind", "source_url", "title", "status", "storage_path", + "bytes", "meta_json", "updated_at"} + fields = [(k, v) for k, v in kwargs.items() if k in allowed] + if not fields: + return + if "updated_at" not in kwargs: + fields.append(("updated_at", time.time())) + if "meta_json" in kwargs and isinstance(kwargs["meta_json"], dict): + # find and replace the meta_json tuple + for i, (k, _v) in enumerate(fields): + if k == "meta_json": + fields[i] = ("meta_json", json.dumps(kwargs["meta_json"])) + break + + set_clause = ", ".join(f"{k} = ?" for k, _ in fields) + values = [v for _, v in fields] + values.append(item_id) + await self._db.execute( + f"UPDATE library_items SET {set_clause} WHERE id = ?", values + ) + await self._db.commit() + + async def delete_item(self, item_id: str) -> None: + """Delete an item and cascade its artifacts and jobs.""" + await self._db.execute("DELETE FROM library_items WHERE id = ?", (item_id,)) + await self._db.commit() + + async def update_item_status(self, item_id: str, status: str) -> None: + if status not in _VALID_STATUSES: + raise ValueError( + f"Invalid status {status!r}; must be one of {sorted(_VALID_STATUSES)}" + ) + await self.update_item(item_id, status=status) + + async def try_update_item_status( + self, item_id: str, new_status: str, *, + if_not_in: tuple[str, ...] = ("pending", "processing"), + ) -> bool: + """Atomically set status to *new_status* when current status is NOT in *if_not_in*. + + Returns True when a row was updated, False otherwise. + Used by reprocess to avoid TOCTOU races — two concurrent reprocess + requests cannot both pass a read-then-write guard. + """ + if new_status not in _VALID_STATUSES: + raise ValueError( + f"Invalid status {new_status!r}; must be one of {sorted(_VALID_STATUSES)}" + ) + placeholders = ",".join("?" * len(if_not_in)) + params = [new_status, time.time(), item_id, *if_not_in] + cursor = await self._db.execute( + f"UPDATE library_items SET status = ?, updated_at = ? " + f"WHERE id = ? AND status NOT IN ({placeholders})", + params, + ) + await self._db.commit() + return cursor.rowcount > 0 + + # -- artifacts -------------------------------------------------------- + + async def add_artifact( + self, item_id: str, kind: str, path: str = "", + meta: dict | None = None, + ) -> str: + artifact_id = uuid.uuid4().hex[:16] + now = time.time() + await self._db.execute( + """INSERT INTO library_artifacts (id, item_id, kind, path, meta_json, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (artifact_id, item_id, kind, path, json.dumps(meta or {}), now), + ) + await self._db.commit() + return artifact_id + + async def get_artifacts(self, item_id: str) -> list[dict]: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_artifacts WHERE item_id = ? ORDER BY created_at", + (item_id,), + ) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + async def delete_artifact(self, artifact_id: str) -> None: + await self._db.execute( + "DELETE FROM library_artifacts WHERE id = ?", (artifact_id,) + ) + await self._db.commit() + + # -- jobs ------------------------------------------------------------- + + async def create_job(self, item_id: str, stage: str) -> str: + job_id = uuid.uuid4().hex[:16] + now = time.time() + await self._db.execute( + """INSERT INTO library_jobs (id, item_id, stage, state, created_at, updated_at) + VALUES (?, ?, ?, 'queued', ?, ?)""", + (job_id, item_id, stage, now, now), + ) + await self._db.commit() + return job_id + + async def get_job(self, job_id: str) -> dict | None: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_jobs WHERE id = ?", (job_id,) + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + + async def get_item_jobs(self, item_id: str) -> list[dict]: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_jobs WHERE item_id = ? ORDER BY created_at", + (item_id,), + ) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + async def update_job(self, job_id: str, **kwargs) -> None: + allowed = {"state", "error", "updated_at"} + fields = [(k, v) for k, v in kwargs.items() if k in allowed] + if not fields: + return + if "updated_at" not in kwargs: + fields.append(("updated_at", time.time())) + + set_clause = ", ".join(f"{k} = ?" for k, _ in fields) + values = [v for _, v in fields] + values.append(job_id) + await self._db.execute( + f"UPDATE library_jobs SET {set_clause} WHERE id = ?", values + ) + await self._db.commit() diff --git a/tinyagentos/projects/invite_store.py b/tinyagentos/projects/invite_store.py index e0c0240df..017c9a63d 100644 --- a/tinyagentos/projects/invite_store.py +++ b/tinyagentos/projects/invite_store.py @@ -10,7 +10,7 @@ from tinyagentos.base_store import BaseStore -_EXPIRY_SECS = 15 * 60 +_EXPIRY_SECS = 60 * 60 _MAX_ATTEMPTS = 5 _PENDING_CAP = 10 @@ -131,7 +131,8 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, display_name: str | None = None, kind: str = "agent", pin_required: bool = True, - contact_id: str | None = None) -> dict: + contact_id: str | None = None, + ttl_secs: int | None = None) -> dict: if self._db is None: raise RuntimeError("ProjectInviteStore not initialised") @@ -180,7 +181,9 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, pin = self._generate_pin() pin_hash = hashlib.sha256(pin.encode()).hexdigest() now = _now() - expires_ts = now + _EXPIRY_SECS + if ttl_secs is not None and ttl_secs <= 0: + raise ValueError("ttl_secs must be positive") + expires_ts = now + (ttl_secs if ttl_secs is not None else _EXPIRY_SECS) # 6-digit invite IDs have a non-trivial collision chance under load # (~12 % at 500 pending). Retry on UNIQUE constraint violation so a @@ -322,7 +325,8 @@ async def revoke(self, invite_id: str) -> bool: if self._db is None: raise RuntimeError("ProjectInviteStore not initialised") cursor = await self._db.execute( - "UPDATE project_invites SET status = 'revoked' WHERE invite_id = ? AND status = 'pending'", + "UPDATE project_invites SET status = 'revoked' " + "WHERE invite_id = ? AND status IN ('pending', 'expired')", (invite_id,), ) await self._db.commit() diff --git a/tinyagentos/routes/__init__.py b/tinyagentos/routes/__init__.py index e0855684d..07e9d7406 100644 --- a/tinyagentos/routes/__init__.py +++ b/tinyagentos/routes/__init__.py @@ -408,6 +408,9 @@ def register_all_routers(app): from tinyagentos.routes.receipts import router as receipts_router app.include_router(receipts_router, dependencies=_csrf) + from tinyagentos.routes.library import router as library_router + app.include_router(library_router, dependencies=_csrf) + from tinyagentos.routes import wallhaven as wallhaven_routes app.include_router(wallhaven_routes.router, dependencies=_csrf) diff --git a/tinyagentos/routes/library.py b/tinyagentos/routes/library.py new file mode 100644 index 000000000..067a98ebc --- /dev/null +++ b/tinyagentos/routes/library.py @@ -0,0 +1,357 @@ +"""Library app routes — ingest, list, and manage library items. + +POST /api/library/ingest — accept file uploads or URL references +GET /api/library/items — list library items +GET /api/library/items/{item_id} — item detail with artifacts +DELETE /api/library/items/{item_id} — remove item and its files +POST /api/library/items/{item_id}/reprocess — re-run the pipeline +""" + +from __future__ import annotations + +import asyncio +import logging +import uuid +from pathlib import Path + +from fastapi import APIRouter, Request, UploadFile, File, Form +from fastapi.responses import JSONResponse + +from tinyagentos.library_pipeline import run_pipeline +from tinyagentos.library_collections import handoff_to_collections +from tinyagentos.task_utils import _create_supervised_task + +logger = logging.getLogger(__name__) + +router = APIRouter() + +LIBRARY_DIR_NAME = "library" + +# Module-level background task tracking so unreferenced tasks are not +# garbage-collected when request.app.state._background_tasks is absent. +_background_tasks: set[asyncio.Task] = set() + + +def _track_background_task(coro) -> asyncio.Task: + """Create a task, store it in ``_background_tasks``, and auto-discard on done.""" + task = asyncio.create_task(coro) + + def _on_done(t: asyncio.Task) -> None: + _background_tasks.discard(t) + + task.add_done_callback(_on_done) + _background_tasks.add(task) + return task + + +# --------------------------------------------------------------------------- + +def _library_dir_from_app(app) -> Path: + """Return the library storage directory, creating it if needed.""" + data_dir = getattr(app.state, "data_dir", None) + if data_dir: + d = Path(data_dir) / LIBRARY_DIR_NAME + else: + d = Path(__file__).parent.parent.parent / "data" / LIBRARY_DIR_NAME + d.mkdir(parents=True, exist_ok=True) + return d + + +def _library_dir(request: Request) -> Path: + return _library_dir_from_app(request.app) + + +async def _get_library_store(request: Request): + """Get the LibraryStore from app.state (lazily initialised).""" + store = getattr(request.app.state, "library_store", None) + if store is None: + from tinyagentos.library_store import LibraryStore + + data_dir = getattr(request.app.state, "data_dir", None) + base = Path(data_dir) if data_dir else Path(__file__).parent.parent.parent / "data" + store = LibraryStore(base / "library.db") + await store.init() + request.app.state.library_store = store + return store + + +# --------------------------------------------------------------------------- +# Ingest +# --------------------------------------------------------------------------- + + +@router.post("/api/library/ingest") +async def ingest( + request: Request, + file: UploadFile | None = File(None), + url: str | None = Form(None), + title: str | None = Form(None), +): + """Ingest a file or URL into the library. + + Accepts a file upload (multipart) or a URL string form field. At least one + of ``file`` or ``url`` must be provided. + + Returns ``{item_id, status: \"pending\"}`` immediately — pipeline processing + happens asynchronously in a background task. + """ + store = await _get_library_store(request) + storage_dir = _library_dir(request) + + if file and file.filename: + # File upload + kind = _detect_kind_from_filename(file.filename, file.content_type) + file_dir = storage_dir / "files" + file_dir.mkdir(parents=True, exist_ok=True) + + # Sanitise filename + safe_name = _sanitise_filename(file.filename) + dest = file_dir / f"{uuid.uuid4().hex[:8]}_{safe_name}" + + # Stream file in bounded chunks to avoid loading it entirely into + # memory. Reject uploads exceeding 100 MB with HTTP 413. + MAX_SIZE = 100 * 1024 * 1024 # 100 MB + if file.size and file.size > MAX_SIZE: + return JSONResponse( + {"error": "Payload Too Large"}, status_code=413, + ) + + size = 0 + with dest.open("wb") as f: + while chunk := await file.read(1024 * 1024): + size += len(chunk) + if size > MAX_SIZE: + dest.unlink(missing_ok=True) + return JSONResponse( + {"error": "Payload Too Large"}, status_code=413, + ) + f.write(chunk) + + item_id = await store.create_item( + kind=kind, + title=title or file.filename, + storage_path=str(dest), + size_bytes=size, + source_url="", + ) + elif url: + # URL reference + kind = _detect_kind_from_url(url) + item_id = await store.create_item( + kind=kind, + source_url=url, + title=title or url, + ) + else: + return JSONResponse( + {"error": "Provide either 'file' (multipart upload) or 'url' (form field)."}, + status_code=400, + ) + + # Run pipeline in background + task_set = getattr(request.app.state, "_background_tasks", None) + coro = _ingest_task(request.app, item_id, store, storage_dir) + if task_set is None: + _track_background_task(coro) + else: + _create_supervised_task(coro, task_set) + + return JSONResponse({"item_id": item_id, "status": "pending"}, status_code=202) + + +async def _ingest_task(app, item_id: str, store, storage_dir: Path) -> None: + """Background task: run pipeline + collections handoff for an item. + + Never raises — always leaves the item in a terminal status. + """ + try: + await run_pipeline(store, item_id, storage_dir) + except Exception: + logger.exception("Library ingest pipeline crashed for item %s", item_id) + await store.update_item_status(item_id, "error") + return + + # Collections handoff after successful pipeline + try: + collections_dir = storage_dir.parent / "collections" + config = getattr(app.state, "config", None) + taosmd_url = getattr(config, "memory_url", None) if config else None + taosmd_admin_token = None + secrets = getattr(app.state, "secrets", None) + if secrets: + secret = await secrets.get("taosmd-admin-token") + if secret: + taosmd_admin_token = secret["value"] + indexed = await handoff_to_collections( + store, item_id, collections_dir, + taosmd_url=taosmd_url, + taosmd_admin_token=taosmd_admin_token, + ) + if indexed > 0: + logger.info( + "Collections handoff indexed %d file(s) for item %s", + indexed, item_id, + ) + else: + logger.debug( + "Collections handoff indexed 0 files for item %s " + "(no text artifacts or taosmd unavailable)", + item_id, + ) + except Exception: + logger.exception("Collections handoff failed for item %s", item_id) + + +def _sanitise_filename(name: str) -> str: + """Strip path separators and null bytes from a filename.""" + name = name.replace("/", "_").replace("\\", "_").replace("\x00", "") + # Also prevent double-dots for path traversal + name = name.replace("..", "_") + return name or "unnamed" + + +def _detect_kind_from_filename(filename: str, content_type: str | None = None) -> str: + """Detect kind from filename and optional MIME type.""" + from tinyagentos.library_pipeline import detect_kind + return detect_kind(file_path=filename, content_type=content_type or "") + + +def _detect_kind_from_url(url: str) -> str: + """Detect kind from URL pattern.""" + from tinyagentos.library_pipeline import detect_kind + return detect_kind(source_url=url) + + +# --------------------------------------------------------------------------- +# List / Get / Delete +# --------------------------------------------------------------------------- + + +@router.get("/api/library/items") +async def list_items( + request: Request, + kind: str | None = None, + status: str | None = None, + limit: int = 50, + offset: int = 0, +): + """List library items, optionally filtered by kind or status.""" + store = await _get_library_store(request) + items = await store.list_items(kind=kind, status=status, limit=limit, offset=offset) + return {"items": items, "count": len(items)} + + +@router.get("/api/library/items/{item_id}") +async def get_item(request: Request, item_id: str): + """Get a library item with its artifacts.""" + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + artifacts = await store.get_artifacts(item_id) + jobs = await store.get_item_jobs(item_id) + return {"item": item, "artifacts": artifacts, "jobs": jobs} + + +@router.delete("/api/library/items/{item_id}") +async def delete_item(request: Request, item_id: str): + """Delete a library item and its associated files.""" + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + # Remove on-disk files + storage_path = item.get("storage_path", "") + if storage_path and (p := Path(storage_path)).exists(): + try: + p.unlink() + except OSError: + logger.warning("Failed to remove storage file %s for item %s", + storage_path, item_id) + + # Remove artifacts from disk + artifacts = await store.get_artifacts(item_id) + for art in artifacts: + art_path = art.get("path", "") + if art_path and (ap := Path(art_path)).exists(): + try: + ap.unlink() + except OSError: + logger.warning("Failed to remove artifact %s for item %s", + art_path, item_id) + + # Remove collections folder + storage_dir = _library_dir(request) + item_collection_dir = storage_dir.parent / "collections" / item_id + if item_collection_dir.exists(): + import shutil + try: + shutil.rmtree(item_collection_dir) + except OSError: + logger.warning("Failed to remove collection dir %s for item %s", + item_collection_dir, item_id) + + await store.delete_item(item_id) + return {"status": "deleted", "item_id": item_id} + + +@router.post("/api/library/items/{item_id}/reprocess") +async def reprocess_item(request: Request, item_id: str): + """Re-run the ingest pipeline for an existing item. + + Idempotent per (item, stage): old artifacts and their on-disk files are + removed before the pipeline re-runs, so no duplicate rows or files accumulate. + """ + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + # Atomic CAS: transition to pending only when NOT already pending/processing. + # Beats the TOCTOU race — two concurrent reprocess requests cannot both + # pass a separate read-then-write guard. + if not await store.try_update_item_status(item_id, "pending", + if_not_in=("pending", "processing")): + return JSONResponse( + {"error": "Item is currently being processed"}, status_code=409 + ) + + # Delete old artifacts so reprocess is idempotent. + # Guard: never unlink the user's original uploaded file (storage_path). + storage_dir = _library_dir(request) + old_artifacts = await store.get_artifacts(item_id) + item_storage_path = item.get("storage_path", "") + for art in old_artifacts: + art_path = art.get("path", "") + if art_path and art_path == item_storage_path: + # This artifact records the source file — skip deletion. + logger.debug("Skipping unlink of source file %s for item %s", + art_path, item_id) + elif art_path and (ap := Path(art_path)).exists(): + try: + ap.unlink() + except OSError: + logger.warning("Failed to remove artifact %s for item %s", + art_path, item_id) + await store.delete_artifact(art["id"]) + + # Status was already atomically set to pending by try_update_item_status above; + # re-queue the pipeline now. If scheduling fails, roll back to ready so the + # item is not stuck pending forever. + try: + task_set = getattr(request.app.state, "_background_tasks", None) + coro = _ingest_task(request.app, item_id, store, storage_dir) + if task_set is None: + _track_background_task(coro) + else: + _create_supervised_task(coro, task_set) + except Exception: + logger.exception("Failed to schedule reprocess for item %s", item_id) + await store.update_item_status(item_id, "ready") + return JSONResponse( + {"error": "Failed to schedule reprocess"}, status_code=500, + ) + + return JSONResponse({"item_id": item_id, "status": "reprocessing"}, status_code=202) diff --git a/tinyagentos/routes/project_invites.py b/tinyagentos/routes/project_invites.py index 8d93caad0..0717c94a6 100644 --- a/tinyagentos/routes/project_invites.py +++ b/tinyagentos/routes/project_invites.py @@ -30,6 +30,10 @@ class MintInviteIn(BaseModel): scopes: list[str] = Field(default_factory=list) approval_mode: str = Field(default="auto", pattern="^(auto|manual)$") check_interval_secs: int = Field(default=1800, ge=60) + # How long the invite stays redeemable. Omitted means the store default + # (1 hour). Capped at 24 hours: the PIN is the security gate, expiry is + # hygiene, but an indefinitely live link is not a state we want. + ttl_secs: int | None = Field(default=None, ge=60, le=86400) class MintOsInviteIn(BaseModel): @@ -605,6 +609,22 @@ def _build_os_guide_markdown( return "\n".join(lines) + +def _scopes_as_list(raw) -> list: + """The store keeps scopes as a JSON-encoded TEXT column; the API contract + is a list. Parse defensively so a malformed row degrades to [] instead of + crashing every client that renders the list.""" + if isinstance(raw, list): + return raw + if isinstance(raw, str): + try: + import json as _json + v = _json.loads(raw) + return v if isinstance(v, list) else [] + except Exception: + return [] + return [] + @router.post("/api/projects/{project_id}/invites") async def mint_invite( project_id: str, @@ -633,6 +653,7 @@ async def mint_invite( approval_mode=payload.approval_mode, check_interval_secs=payload.check_interval_secs, created_by=user.user_id, + ttl_secs=payload.ttl_secs, ) except InvitePendingCapError as exc: return JSONResponse({"error": str(exc)}, status_code=429) @@ -663,7 +684,7 @@ async def list_invites( return [ { "invite_id": i["invite_id"], - "scopes": i["scopes"], + "scopes": _scopes_as_list(i["scopes"]), "status": i["status"], "expires_ts": i["expires_ts"], "redeemed_by": i.get("redeemed_by"), @@ -688,9 +709,19 @@ async def revoke_invite( row = await store.get(invite_id) if row is None or row.get("project_id") != project_id: return JSONResponse({"error": "invite not found"}, status_code=404) + status = row.get("status") + if status in ("redeemed", "revoked"): + return JSONResponse({"error": f"invite already {status}"}, status_code=409) + # 'claimed' is the transient mid-redeem state. Revoking it would race the + # redeem, so refuse with a message that names the state rather than the + # generic retry text. + if status == "claimed": + return JSONResponse( + {"error": "invite is mid-redeem, retry once it settles"}, status_code=409 + ) ok = await store.revoke(invite_id) if not ok: - return JSONResponse({"error": "invite not found or already redeemed"}, status_code=404) + return JSONResponse({"error": "invite state changed, retry"}, status_code=409) return JSONResponse(content=None, status_code=204) @@ -768,7 +799,7 @@ async def list_os_invites( return [ { "invite_id": i["invite_id"], - "scopes": i["scopes"], + "scopes": _scopes_as_list(i["scopes"]), "status": i["status"], "expires_ts": i["expires_ts"], "redeemed_by": i.get("redeemed_by"), diff --git a/uv.lock b/uv.lock index 11a077570..677bb2920 100644 --- a/uv.lock +++ b/uv.lock @@ -3022,7 +3022,7 @@ wheels = [ [[package]] name = "tinyagentos" -version = "1.0.0b42" +version = "1.0.0b43" source = { editable = "." } dependencies = [ { name = "aiosqlite" },