diff --git a/.env.example b/.env.example index d8be53e..c124aee 100644 --- a/.env.example +++ b/.env.example @@ -42,8 +42,12 @@ FRONTEND_PORT=8642 HF_TOKEN= # LLM provider profile — which LiteLLM backend the agent talks to. -# Options are defined in braindb/config.py::_LLM_PROFILES -# (currently: nim, deepinfra, openai_compatible, vllm_workstation). +# Options are defined in braindb/config.py::_LLM_PROFILES — currently: +# hosted : nim, deepinfra +# bring-your-own: openai_compatible (set OPENAI_BASE_URL + AGENT_MODEL) +# self-hosted : vllm_workstation, vllm_workstation_qwen, +# vllm_workstation_gemma, vllm_workstation_gemma12b +# The vllm_* profiles carry a fixed host/port; check they match your server. LLM_PROFILE=deepinfra # Provider API keys — fill in whichever profile you're using. @@ -98,6 +102,27 @@ AGENT_VERBOSE=false # Layer 4 retry path). # AGENT_COUNTDOWN_THRESHOLD=8 +# Per-LLM-call HTTP deadline in seconds, default 4800 (80 min). Passed to +# LiteLLM as `timeout=`. Without an explicit value LiteLLM falls back to +# 600s, which is long enough for hosted providers but NOT for a self-hosted +# quantised 27B doing a full wiki write — the client abandoned a request the +# server was still completing, so the work was computed and thrown away. +# This is a ceiling, not a delay: fast providers finish far inside it and are +# unaffected. NOTE: setting this to exactly 6000 is a no-op (that value is +# LiteLLM's own sentinel). +# AGENT_REQUEST_TIMEOUT=4800 + +# Reasoning effort for the WIKI agents only (maintainer / writer / +# subagent). Blank = send nothing = the server's own default. On the Qwen3 +# chat template that default is 'xhigh' (its maximum), so setting this to +# 'low' is a large latency win; the SDK discards reasoning between turns on +# non-DeepSeek/Claude models, so nothing is lost across turns. Valid on that +# template: low | medium | none. NOT minimal/high — the template raises. +# The general agent (/agent/query + ingest watcher) is deliberately unaffected. +# SELF-HOSTED vLLM ONLY — it rides in the body as chat_template_kwargs, which +# a hosted provider may reject. Leave blank on the deepinfra/nim profiles. +# AGENT_WIKI_REASONING_EFFORT= + # Ingest watcher poll interval (seconds) — how often the watcher sidecar # scans data/sources/ for new files to ingest. INGEST_POLL_INTERVAL=7 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index fd22ed5..4f5e52d 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -67,11 +67,19 @@ jobs: - name: Install pytest into the api container run: docker exec braindb_api pip install pytest pytest-asyncio --quiet - - name: Run validator + handoff unit tests + - name: Run validator + handoff + wiki-writer guard tests + # The wiki files below are the revert-detectors for the writer-loop + # fixes: sections/guards are DB-free; selfheal + reconcile_dangling + # use the workflow's Postgres service via the container's + # DATABASE_URL (schema comes from the api's alembic upgrade). run: | docker exec braindb_api python -m pytest \ tests/test_final_answer_rename.py \ tests/test_handoff_hooks.py \ + tests/test_wiki_sections.py \ + tests/test_wiki_writer_guards.py \ + tests/test_wiki_reconcile_dangling.py \ + tests/test_wiki_selfheal_db.py \ -v - name: Dump api logs on failure diff --git a/.gitignore b/.gitignore index 4fd6851..1627ee0 100644 --- a/.gitignore +++ b/.gitignore @@ -62,6 +62,18 @@ data_bench/ # Test stack (docker-compose.test.yml): separate host data dir for test ingests data_test/ +# Integration runtime state — per-caller conversation transcripts and other +# local state written by integrations at run time. Never belongs in the repo. +integrations/*/.state/ + +# Local operational artefacts. DB dumps, logs and editor backups have no place +# in a public repo and are easy to stage by accident with `git add .`. +*.log +*.bak +*.sql +!scripts/*.sql +backups/ + # Hermes sandbox (integrations/hermes/sandbox): throwaway agent profile dir — # holds a .env with the LLM key + provider state; never commit it. integrations/hermes/sandbox/hermes-data/ diff --git a/CHANGELOG.md b/CHANGELOG.md index c0bf9de..6fbf87e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,75 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.10.0] — 2026-09-12 + +Headline: **the wiki pipeline is now dependable under long, unattended runs.** The writer loop +always terminates, crashed and restarted jobs recover on their own, and the page header is editable +instead of frozen after the first write. Also in this release: a reasoning-effort knob for the wiki +agents that cuts latency several-fold on self-hosted models, a per-call LLM timeout so slow local +writes are no longer abandoned mid-flight, and both agents can now see how big a page is before +deciding what to do with it. + +### Added + +- **Append mode and paging for the wiki section tools.** `edit_wiki_section` gains `mode="append"`, + and `read_wiki_section` takes `offset`/`limit`. Reads were capped well below the uncapped write + size, so a writer could be asked to preserve a section it was only ever shown part of — an + append-shaped job forced through a replace-shaped tool. The model now picks the edit that fits the + content rather than the one the tool allows. +- **`check_members_cited`.** One shared predicate answering "is this member cited yet?", used by + both the writer's tool and the router's gate, so the two cannot disagree. Three copies of that + logic previously existed and had drifted. +- **The page header is an editable section.** Meta line, title, Summary and Disambiguation were + effectively frozen once a page grew past the inline-body limit. They are now readable and + replaceable through the existing section tools, so a page whose story changes can have its opening + changed too. +- **`AGENT_WIKI_REASONING_EFFORT`.** Wiki agents only, blank by default. Some chat templates default + reasoning to their maximum when no value is sent, and the SDK discards that reasoning between + turns on most model families — so it is generated, paid for, and dropped. Self-hosted vLLM only; + leave blank on hosted providers. +- **`AGENT_REQUEST_TIMEOUT`.** The per-call transport deadline, default 4800s. +- **Size awareness for both agents.** The maintainer sees each page's size in its catalog; the + writer sees neighbouring page names and sizes. Neither could previously tell a large page from an + empty one, so every candidate target looked equally reasonable. + +### Changed + +- **Writer handoff budget raised 20000 -> 30000.** A budget set where it cannot fire silently + disables the successor path, leaving long writes to grow until they hit the turn limit instead of + handing off to a fresh successor. +- **Job lease raised 20 -> 120 min, with a bounded reclaim ceiling.** A long write is no longer + mistaken for an abandoned one. +- **`vllm_workstation_qwen` profile** now points at the Qwen model and port the wiki pipeline is + actually tuned against, so selecting it needs no `AGENT_MODEL` override. `deepinfra` remains the + default profile. + +### Fixed + +- **The writer loop now terminates.** A long unattended run spent most of its time on a single page + whose work was already complete — every member was already cited, yet the job kept being re-run. + Cause was the tool mismatch above plus a reconcile step that raised on a stale reference instead + of skipping it, aborting the very transaction that would have closed the job. +- **Self-healing restored.** Jobs past their lease and reclaim ceiling now fail and re-enter triage + in the same sweep instead of wedging indefinitely. An entity can no longer be silently lost. +- **Crashes and restarts are recoverable.** The pre-write snapshot is taken when the job is claimed + rather than after the model runs, so an interrupted run is always reversible; jobs orphaned by a + restart are returned to the queue on startup. +- **Token estimate counts tool results.** It read only message content, missing the tool-result + payloads that dominate a writer's context — so the handoff nudge never fired at any budget. +- **Per-call LLM timeout.** LiteLLM's own 600s client fallback was abandoning self-hosted writes the + server was still completing, so the work was computed and discarded. Hosted providers finish well + inside the new ceiling and are unaffected — it is a ceiling, not a delay. +- **Duplicate page creation.** Two create jobs for the same proposed name within the same window now + collapse to one. +- **`update_entity` no longer overwrites wiki bodies**, and a blank body is a warned no-op rather + than a silent wipe. Subagents gained the wiki READ tools so they no longer fall back to retyping a + body they can only partly see. + +### Upgrading from v0.9.0 + +No DB migration and no required env changes. Both new knobs default to the previous behaviour. + ## [0.9.0] — 2026-06-26 Headline: **custom profiles** — opt-in, self-contained overlays that reshape what BrainDB ingests diff --git a/CLAUDE.md b/CLAUDE.md index d4567dd..a5671c1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -193,7 +193,7 @@ When debugging the agent: set `AGENT_VERBOSE=true` in `.env` and watch `docker l ## Important Notes -- `.env` contains real DB credentials and provider API keys (`DEEPINFRA_API_KEY`, `NVIDIA_NIM_API_KEY`, etc.) — **never commit it**, it is in `.gitignore`. Active provider is picked by `LLM_PROFILE` (see `braindb/config.py::_LLM_PROFILES`). `LLM_PROFILE=deepinfra` (model `google/gemma-4-31B-it`) is the recommended starting point — fast, cheap, validated end-to-end; the `vllm_*` profiles are for advanced/offline use and need a workstation GPU + SSH tunnel. +- `.env` contains real DB credentials and provider API keys (`DEEPINFRA_API_KEY`, `NVIDIA_NIM_API_KEY`, etc.) — **never commit it**, it is in `.gitignore`. Active provider is picked by `LLM_PROFILE` (see `braindb/config.py::_LLM_PROFILES`). `LLM_PROFILE=deepinfra` (model `google/gemma-4-31B-it`) is the recommended starting point — fast, cheap, validated end-to-end; the `vllm_*` profiles need a GPU serving an OpenAI-compatible endpoint reachable from the docker network. Note the wiki pipeline's tuning constants (turn budget, request timeout, writer handoff budget, `AGENT_WIKI_REASONING_EFFORT`) were measured against self-hosted Qwen on vLLM — `vllm_workstation_qwen` is that profile. - Always-on rules (priority 100, `always_on: true`) are returned on every `/memory/context` call - `notes` field on any entity or relation is for running commentary — append observations over time - Keywords are stored as both a `TEXT[]` column on the entity AND as separate keyword entities linked via `tagged_with` relations (the keyword entities carry the embeddings for semantic search) diff --git a/braindb/agent/agent.py b/braindb/agent/agent.py index 34d7769..57277e5 100644 --- a/braindb/agent/agent.py +++ b/braindb/agent/agent.py @@ -22,6 +22,7 @@ """ import json import logging +from uuid import uuid4 from pathlib import Path from typing import TypeVar @@ -31,7 +32,13 @@ from pydantic import BaseModel from braindb.agent.hooks import CountdownHooks -from braindb.agent.run_state import install_slot, release_slot +from braindb.agent.run_state import ( + get_run_tag, + install_slot, + release_slot, + reset_run_tag, + set_run_tag, +) from braindb.agent.schemas import ( AgentAnswer, MaintainerDecision, @@ -39,6 +46,7 @@ WikiWriteResult, ) from braindb.agent.tools import ( + check_members_cited, create_relation, delegate_to_subagent, delete_entity, @@ -169,6 +177,7 @@ def _build( submit_tool, extra_tools: tuple = (), extra_stop_tools: tuple[str, ...] = (), + reasoning_effort: str = "", ) -> Agent: """Build an agent. NOTE: no `output_type` — see module docstring. The structured contract lives on `submit_tool`'s argument schema, not on @@ -181,13 +190,43 @@ def _build( `extra_stop_tools` adds extra stop-tool names beyond `final_answer`. The writer adds `handoff_to_successor` here so the run halts cleanly when handoff is called instead of continuing wastefully. + + `reasoning_effort` (blank = send nothing) is passed only by the wiki + agents; see `settings.agent_wiki_reasoning_effort`. """ set_tracing_disabled(disabled=True) + # `extra_args` is forwarded verbatim into the LiteLLM call. `timeout` + # lands as `kwargs["timeout"]` — which LiteLLM resolves ahead of its + # 600s fallback; a TRANSPORT deadline only, it never enters the request + # body and cannot steer the model, unlike `output_type` / `tool_choice` + # (see the module docstring — those stay unset deliberately). Without it + # a long wiki write is abandoned client-side at 600s while the server is + # still working. + # + # `reasoning_effort` belongs in the request BODY, but it cannot travel as + # a plain kwarg: the SDK's LitellmModel lifts that exact key out of + # `reasoning` / `extra_body` / `extra_args` and promotes it to a top-level + # `reasoning_effort=` argument on `litellm.acompletion()`, where LiteLLM + # checks it against a PER-PROVIDER allow-list — `openai` (which every + # OpenAI-compatible profile resolves to) does not list it, so the call + # raises `UnsupportedParamsError` before any request is sent. Nesting it + # under `chat_template_kwargs` sidesteps that: the SDK only intercepts the + # literal top-level key, so the dict is copied through into LiteLLM's + # `extra_body` and forwarded into the JSON body unfiltered, which is where + # vLLM reads chat-template variables from. VLLM-SPECIFIC by nature — a + # hosted provider may reject the unknown body key, so this stays blank + # unless an operator opts in on a self-hosted profile. + extra_args = {"timeout": settings.agent_request_timeout} + extra_body = ( + {"chat_template_kwargs": {"reasoning_effort": reasoning_effort}} + if reasoning_effort else None + ) agent = Agent( name=name, instructions=SYSTEM_PROMPT, model=_model(), - model_settings=ModelSettings(), + model_settings=ModelSettings(extra_args=extra_args, + extra_body=extra_body), tools=[*_BASE_TOOLS, *extra_tools, submit_tool], tool_use_behavior=StopAtTools( stop_at_tool_names=["final_answer", *extra_stop_tools], @@ -209,6 +248,7 @@ def _cached( submit_tool, extra_tools: tuple = (), extra_stop_tools: tuple[str, ...] = (), + reasoning_effort: str = "", ) -> Agent: a = _cache.get(key) if a is None: @@ -216,6 +256,7 @@ def _cached( name, submit_tool, extra_tools=extra_tools, extra_stop_tools=extra_stop_tools, + reasoning_effort=reasoning_effort, ) _cache[key] = a return a @@ -230,6 +271,7 @@ def _cached( _WRITER_EXTRA_TOOLS = ( read_wiki_outline, read_wiki_section, + check_members_cited, edit_wiki_section, delete_wiki_section, validate_wiki, @@ -237,6 +279,21 @@ def _cached( ) _WRITER_EXTRA_STOP_TOOLS = ("handoff_to_successor",) +# Subagent extras: the wiki READ tools, and nothing that writes. A writer +# routinely delegates "check/read this page" work, and without these the +# subagent could not do it the safe way — it fell back to paging the raw body +# and re-emitting it through `update_entity`, which is both enormously +# expensive and how a cited UUID gets corrupted. Giving it the read tools +# removes the reason to do that. Edit/delete stay writer-only on purpose: +# one writer per wiki keeps the revision CAS meaningful, and a subagent +# cannot hand off, so it has no business holding a revision token. +_SUBAGENT_EXTRA_TOOLS = ( + read_wiki_outline, + read_wiki_section, + check_members_cited, + validate_wiki, +) + def get_agent() -> Agent: """Default agent: general recall/save (public /agent/query).""" @@ -244,7 +301,8 @@ def get_agent() -> Agent: def get_maintainer_agent() -> Agent: - return _cached("maintainer", "BrainDB Wiki Maintainer", submit_maintainer) + return _cached("maintainer", "BrainDB Wiki Maintainer", submit_maintainer, + reasoning_effort=settings.agent_wiki_reasoning_effort) def get_writer_agent() -> Agent: @@ -252,11 +310,17 @@ def get_writer_agent() -> Agent: "writer", "BrainDB Wiki Writer", submit_wiki, extra_tools=_WRITER_EXTRA_TOOLS, extra_stop_tools=_WRITER_EXTRA_STOP_TOOLS, + reasoning_effort=settings.agent_wiki_reasoning_effort, ) def get_subagent() -> Agent: - return _cached("subagent", "BrainDB Subagent", submit_subagent) + # Shared surface: any agent can delegate, so a subagent spawned from + # /agent/query also inherits the wiki effort setting. Accepted because + # subagent runs are overwhelmingly wiki work. + return _cached("subagent", "BrainDB Subagent", submit_subagent, + extra_tools=_SUBAGENT_EXTRA_TOOLS, + reasoning_effort=settings.agent_wiki_reasoning_effort) def create_braindb_agent() -> Agent: @@ -292,6 +356,10 @@ async def run_typed( """ turns = max_turns or settings.agent_max_turns slot, token = install_slot() + # Short log tag for THIS run, inherited by the SDK's child Tasks so + # every TOOL line it emits is attributable (see run_state.set_run_tag). + # Set before Runner.run — ContextVars are captured at Task creation. + tag_token = set_run_tag(uuid4().hex[:6]) # Layer-3 nudge: when the run is about to exhaust `max_turns`, the hook # appends a synthetic "you have N turns left, finalise via final_answer" # user message to the conversation. One nudge per run; disabled when @@ -308,7 +376,8 @@ async def run_typed( handoff_tool_name="handoff_to_successor", ) try: - logger.info("Running typed query (%s): %s", agent.name, query[:160]) + logger.info("Running typed query [%s] (%s): %s", + get_run_tag(), agent.name, query[:160]) result = await Runner.run( starting_agent=agent, input=query, max_turns=turns, hooks=hooks, ) @@ -424,6 +493,7 @@ async def run_typed( _bad_request_retried=True, ) finally: + reset_run_tag(tag_token) release_slot(token) diff --git a/braindb/agent/hooks.py b/braindb/agent/hooks.py index 8d1fa9b..bdb760b 100644 --- a/braindb/agent/hooks.py +++ b/braindb/agent/hooks.py @@ -48,17 +48,33 @@ def _estimate_tokens(input_items: list) -> int: - `{"role": str, "content": str}` (LiteLLM dict form) - `{"role": str, "content": [{"type":"text","text":str}, ...]}` (some providers send a list of parts) - - SDK item objects with a `.content` attribute - Unknown shapes contribute 0; the estimate is a lower bound, which - is the safe side for "is context filling up" decisions (we'd rather - fire the handoff nudge slightly late than slightly never).""" + - `{"type": "function_call_output", "output": str}` — a tool result + keeps its payload under `output`, never `content`. These dominate + a writer's context (section reads, recalls), so reading only + `content` left the estimate a near-constant and the handoff nudge + never fired at any budget. + - `{"type": "function_call", "arguments": str}` — an assistant TOOL + CALL carries its payload under `arguments`, with neither `content` + nor `output`. For a writer this is the single largest term in the + conversation (an `edit_wiki_section` call carries a whole section), + and it was previously counted as zero. + - SDK item objects with a `.content`, `.output` or `.arguments` attr + Unknown shapes contribute 0. Deliberately excluded: the system prompt + and the tool schemas, which the SDK passes outside `input_items`. Both + are CONSTANT across a run, so they shift the threshold but never the + growth this watches. The estimate stays a lower bound, which is the + safe side for "is context filling up" decisions (we'd rather fire the + handoff nudge slightly late than slightly never).""" total_chars = 0 for item in input_items: content: object if isinstance(item, dict): - content = item.get("content", "") + content = (item.get("content") or item.get("output") + or item.get("arguments") or "") else: - content = getattr(item, "content", "") + content = (getattr(item, "content", None) + or getattr(item, "output", None) + or getattr(item, "arguments", "")) if isinstance(content, str): total_chars += len(content) elif isinstance(content, list): diff --git a/braindb/agent/prompts/system_prompt.md b/braindb/agent/prompts/system_prompt.md index 1bcb891..78a42b9 100644 --- a/braindb/agent/prompts/system_prompt.md +++ b/braindb/agent/prompts/system_prompt.md @@ -63,8 +63,9 @@ fall back to flat SQL. to see narrative chains. 3. **`delegate_to_subagent`** — for any multi-step investigation or disambiguation ("is this the same person/thing?", "find and resolve X"). - A fresh agent with the full toolset; returns a summary. Prefer this over - doing a long crawl yourself. + A fresh agent with the memory tools (saves and relations included) and + the wiki READ tools; it has no wiki WRITE tools. Returns a summary. + Prefer this over doing a long crawl yourself. 4. `view_entity_relations` / `get_entity` / `list_entities` — direct lookups (single-hop relations, full body of one entity, listing by filter). 5. **`search_sql` ⚠ exception only — aggregates only** (counts, GROUP BY, @@ -94,7 +95,9 @@ design — research from previews, then open only the few you actually need. ## DELEGATION — use `delegate_to_subagent` for focused deep work -When a task would require many tool calls (deep search, duplicate detection, bulk relation work, graph exploration) and you don't need to see the intermediate results in your own context, delegate it to a subagent. The subagent runs in its own conversation context, uses the same tools you have, and returns only a final summary. +When a task would require many tool calls (deep search, duplicate detection, bulk relation work, graph exploration) and you don't need to see the intermediate results in your own context, delegate it to a subagent. The subagent runs in its own conversation context and returns only a final summary. + +**What it can and cannot do.** It has the memory tools — saves, relations, `update_entity`, the full research set — plus the wiki READ tools (`read_wiki_outline`, `read_wiki_section`, `check_members_cited`, `validate_wiki`). It has NO wiki write tools and no handoff. Delegate investigation, reading and ordinary memory work — never a WIKI edit. Asking it to "edit section X" wastes a whole run: it has no safe way to do it. **Write the task description carefully** — the subagent doesn't see your prior conversation, only the task string you pass. Include: - The specific goal diff --git a/braindb/agent/prompts/wiki_maintainer_prompt.md b/braindb/agent/prompts/wiki_maintainer_prompt.md index 6fa7d79..27932e7 100644 --- a/braindb/agent/prompts/wiki_maintainer_prompt.md +++ b/braindb/agent/prompts/wiki_maintainer_prompt.md @@ -78,6 +78,9 @@ attach/consolidate to wikis that appear in that numbered catalog. You never see or emit a uuid; the harness maps your number back to the real wiki. If the subject is not in the catalog, you cannot attach/consolidate to it. +Each catalog entry ends with its current size, e.g. `3. (48210ch)`. +Use it when choosing between `attach` and `create` — see step 4. + ## Decide ONE action PER SEED — STRICT PRECEDENCE, in this order Evaluate top to bottom and take the FIRST that applies. `create` is the last @@ -94,11 +97,20 @@ honour it. subject). Put their catalog **numbers** in `consolidate_nos` (≥2). Do NOT re-propose a pair already linked by `not_duplicate` / `duplicate_of`. This is the primary heal action — if you see duplicates in the catalog - while researching, you MUST propose this. + while researching, you MUST propose this. A page that is narrow but + coherent — its own person, project or event — is NOT a fragment of the + larger page that mentions it; leave it standing. 4. **attach** — a catalog wiki already covers this subject (under any name variant), or the seed is a narrow fact about an already-wikied broad subject. Put that wiki's catalog **number** in `target_wiki_no`. A narrow fact about an existing subject is ALWAYS an attach, never a new page. + **Size exception**: if that target is already very large (roughly 30000ch + or more, per the catalog) AND this seed is a coherent narrower subject in + its own right — a person, project or event the page keeps referring to — + prefer `create` for that narrower subject instead. It is an ordinary page, + just scoped tighter, so one page does not end up absorbing everything. + This applies only to a genuine standalone subject: a loose fact with no + subject of its own still attaches. 5. **create** — ONLY if steps 1-4 do not apply: recall + the catalog genuinely show no existing wiki for this subject under any variant, AND the evidence supports a clear, explicitly-named subject and scope. Give diff --git a/braindb/agent/prompts/wiki_writer_prompt.md b/braindb/agent/prompts/wiki_writer_prompt.md index 31ea663..2d9b32f 100644 --- a/braindb/agent/prompts/wiki_writer_prompt.md +++ b/braindb/agent/prompts/wiki_writer_prompt.md @@ -24,6 +24,26 @@ claim carries an inline reference `[[ref:ENTITY_UUID]]` (optionally ### Current wiki body (attach mode; empty otherwise) %%CURRENT_BODY%% +### Neighbouring pages (subjects one hop from these members) +%%RELATED_WIKIS%% + +These already exist and cover their own subjects. When a detail belongs to one +of them, **name that page in prose** rather than restating its content here. +Refer to them by NAME only — do NOT +`[[ref:]]` them; refs are for the source entities this page cites. + +Do NOT read a neighbouring page. The name and size above are all you need to +decide whether a detail belongs elsewhere; opening one costs context for no +gain. + +If no listed page fits a detail you turned up while researching — something +that is NOT one of this job's MEMBERS — leave it out. It stays an orphan and +comes back for its own page later; that is the normal path, not a failure. + +**This never applies to a MEMBER of this job.** A dropped member is not +re-queued — the write records it as covered either way — so dropping one loses +it silently. Every member must be cited; see "Citation is mechanical" below. + ### Duplicate wikis to consolidate (consolidate mode only — NUMBERED; pick the survivor's number as `canonical_no`) %%DUPLICATES%% @@ -147,7 +167,7 @@ may leave the run un-cited**. ## Recommended structure (consistency, not a hard gate) ``` - + # NAME > **Summary:** one tight line (aim <= 280 chars) > **Disambiguation:** what this is / is NOT; distinguish it from similarly @@ -158,7 +178,9 @@ may leave the run un-cited**. narrative provenance one bullet per distinct [[ref:UUID]] you cited, with a short note — YOU author this to match - your inline citations + your inline citations, and you may compact or + merge its bullets over time (it is a ledger, + not claims; see the references exception below) ``` `keywords=` in the meta line is optional — list the concept terms that best @@ -177,33 +199,72 @@ can exhaust the context window. Use the section-edit tools instead — they let you read the OUTLINE only (cheap) and rewrite one section at a time, persisting each change immediately: +- `check_members_cited(wiki_id, entity_ids)` — **call this FIRST.** It + answers, exactly and in one call, which of your MEMBERS the page + already cites. If none are missing, the page already covers this job: + verify nothing else needs correcting, then finish with + `final_answer(mode="attach", body="")`. It answers COVERAGE only — not + placement or phrasing. You still read any section you intend to change. - `read_wiki_outline(wiki_id)` — section names + char counts + the - current `revision` token. ALWAYS call this first. -- `read_wiki_section(wiki_id, section_name)` — fetch one section's - content + revision. Read only the section(s) you need to touch. -- `edit_wiki_section(wiki_id, section_name, new_content, expect_revision)` - — replace a section, or append a new one if `section_name` doesn't - exist yet. Pass the latest revision you read; on mismatch you get a - "stale revision" error and must re-read before retrying. + current `revision` token. Call this before any edit. +- `read_wiki_section(wiki_id, section_name, offset, limit)` — fetch one + section + revision. **Read the section you are about to change** — + where new material belongs, and how, is your judgement, and you cannot + judge what you have not read. A section larger than one slice is + paged, not cut: follow `content_meta.next_offset` until it is null + when you need all of it. +- `edit_wiki_section(...)` — `mode="replace"` (the default) rewrites the + section: read it in full first, because anything you do not re-emit is + gone. `mode="append"` adds your text at the end and preserves + everything already there. **Choose by CONTENT, not by cost:** + - The member **corroborates or refines a claim the section already + makes** → integrate it: revise that sentence and stack the citation + (`[[ref:existing]][[ref:new]]`). Never restate as new what the page + already says — a duplicate sentence is worse than a stacked ref. + - The member is **genuinely new information** → append it, or place it + where it reads naturally via replace if the end is the wrong spot. + - The **section's story has changed** (a contradiction resolved, an + event superseded — e.g. an application that became an accepted + offer) → rewrite the section so it tells one story. That freedom is + yours; a snapshot taken when your run claimed this job makes the page + reversible to this run's starting revision. - `delete_wiki_section(wiki_id, section_name, expect_revision)` — remove a section. - `validate_wiki(wiki_id)` — check refs resolve and grammar invariants hold. Run after a batch of edits to catch any broken `[[ref:UUID]]`. +**After your edit, the section must read as one coherent narrative.** It +must never, for example, say the user is applying for a job that a later +line says they already accepted. When you do rewrite, copy `[[ref:UUID]]` +tokens exactly — retyping a UUID by hand is how a digit flips and a +citation dies. + Section-edit grammar invariants when you author `new_content`: - Inline citations stay `[[ref:UUID]]` or `[[ref:UUID|display]]` (grouped form `[[ref:UUID1], [ref:UUID2]]` is also tolerated). - DO NOT include the `` marker yourself — the tool emits it. Your `new_content` is the section's text only. - The HEADER (meta line, `# Title`, `> **Summary:**` / - `> **Disambiguation:**`) lives ABOVE the first section marker. - Section edits never touch the header — if the summary needs to - change, either re-edit the `overview` section to reflect the new - scope, or fall back to a full-body rewrite. + `> **Disambiguation:**`) lives ABOVE the first section marker and is + editable as the reserved section `"header"` — replace-only: read it + (`read_wiki_section(wiki_id, "header")`), then re-emit the whole + block via `edit_wiki_section(wiki_id, "header", ..., mode="replace")`. + Keep the `` line (it is where keywords come + from), and drop any stale `revision=` token — the database owns the + revision. **Update the header whenever the page's story changes**: a + Summary asserting what the body now records differently is a + coherence defect, and the header is what readers see first. - The "Preserve prior work" rule above applies PER SECTION: a replaced section's `new_content` must include every still-valid prior claim + `[[ref:UUID]]` from that section, plus the new - material — a superset, not a lossy summary. + material — a superset, not a lossy summary. This is why you must page + a large section to the end before replacing it (append satisfies the + rule by construction, but choose the edit by content, not by cost). + **One scoped exception — the `references` section**: it is a + bookkeeping ledger, not claims. You may compact it — merge bullets, + drop redundant ones — provided every previously-cited UUID keeps at + least one inline `[[ref:UUID]]` citation somewhere on the page. + Relations are additive, so compaction has no destructive side-effect. When finished, call `final_answer` with `body=""` (empty string) and `mode="attach"`. The router detects that the wiki's revision advanced diff --git a/braindb/agent/run_state.py b/braindb/agent/run_state.py index 0dd6fd5..79bf005 100644 --- a/braindb/agent/run_state.py +++ b/braindb/agent/run_state.py @@ -60,6 +60,32 @@ def release_slot(token: object) -> None: _slot_var.reset(token) # type: ignore[arg-type] +# Run tag — a short id `run_typed` sets for the duration of ONE agent run, +# read DOWNWARD by tool logging (`_verbose` in tools.py) inside the SDK's +# child Tasks, so concurrent runs' TOOL lines are separable in `docker +# logs`. Same read-down pattern as the delegation-depth ContextVar in +# tools.py: no mutable slot needed, nothing ever writes it back up. A +# nested run (delegate_to_subagent -> run_typed) sets its own tag; the +# delegate call itself is logged under the parent's tag, which is the +# correlation point. (Audits of concurrent writer+maintainer+subagent +# logs previously had to attribute calls by argument fingerprint.) +_run_tag_var: ContextVar[str] = ContextVar("braindb_run_tag", default="") + + +def set_run_tag(tag: str) -> object: + """Set the current run's log tag; returns the reset token.""" + return _run_tag_var.set(tag) + + +def reset_run_tag(token: object) -> None: + _run_tag_var.reset(token) # type: ignore[arg-type] + + +def get_run_tag() -> str: + """The current run's log tag, or "" outside any run.""" + return _run_tag_var.get() + + def record_submit(payload: Any) -> None: """Called from inside every `submit_*` tool body. The SDK has already validated `payload` against the tool's Pydantic argument schema, so diff --git a/braindb/agent/tools.py b/braindb/agent/tools.py index 287603d..cd24f44 100644 --- a/braindb/agent/tools.py +++ b/braindb/agent/tools.py @@ -16,6 +16,7 @@ import json import logging import time +from contextvars import ContextVar from typing import Optional from uuid import UUID @@ -36,7 +37,8 @@ ) from braindb.services.search import fuzzy_search, preview, slice_content from braindb.services import wiki_sections as ws -from braindb.agent.run_state import record_handoff, record_submit +from braindb.services import wiki_jobs as wj +from braindb.agent.run_state import get_run_tag, record_handoff, record_submit from braindb.agent.schemas import ( AgentAnswer, MaintainerClusterDecision, @@ -49,6 +51,26 @@ MAX_OUTPUT_CHARS = 8000 +# Cap on the args/result previews `_verbose` writes to the log. 500 proved +# too small to audit real runs (briefs and returns were unreadable stubs); +# this is a LOG cap only, unrelated to MAX_OUTPUT_CHARS (the tool-payload +# cap the model sees). +VERBOSE_PREVIEW_CHARS = 1500 + +# Entity types whose BODY is not editable through the generic `update_entity` +# path, mapped to the redirect the model should follow instead. A wiki body is +# owned by the section tools: they carry the revision CAS, the pre-write +# snapshot and the `summarises` reconcile, and a bare overwrite here bypasses +# all three. It is also how a cited UUID loses a digit — retyping a 50k-char +# body by hand to change one line — after which every later write on that page +# fails on a dangling reference. +_CONTENT_READONLY = { + "datasource": "datasource bodies are read-only; use notes for analysis", + "wiki": 'wiki bodies are owned by the section tools; use ' + 'edit_wiki_section(..., mode="append") to add, or mode="replace" ' + 'to rewrite a section you have read in full', +} + def _truncate(s: str) -> str: if len(s) > MAX_OUTPUT_CHARS: @@ -61,6 +83,18 @@ def _err(msg: str) -> str: return f"ERROR: {msg}" +def _wiki_not_found(wiki_id: str) -> str: + """One message for every wiki tool, stating the three possible causes and + the general recovery. Models mistype UUIDs (observed: splicing one id's + tail onto another's head); a bare "not found" leaves them guessing, while + "copy it exactly" is the recovery that works for any typo shape.""" + return _err( + f"wiki not found: {wiki_id} — either no entity has this id, the " + f"entity is not a wiki, or it has no wiki record. Copy the wiki id " + f"from your job prompt exactly; do not retype UUIDs." + ) + + def _verbose(name: str): """Decorator that logs tool entry and exit when settings.agent_verbose is True. Placed BELOW @function_tool so the SDK still introspects the real signature. @@ -83,21 +117,24 @@ async def wrapper(*args, **kwargs): bound[param_names[i]] = val bound.update(kwargs) try: - args_preview = json.dumps(bound, default=str)[:500] + args_preview = json.dumps(bound, default=str)[:VERBOSE_PREVIEW_CHARS] except Exception: - args_preview = str(bound)[:500] - logger.info("TOOL %s args=%s", name, args_preview) + args_preview = str(bound)[:VERBOSE_PREVIEW_CHARS] + logger.info("TOOL [%s] %s args=%s", + get_run_tag() or "-", name, args_preview) t0 = time.perf_counter() try: result = await fn(*args, **kwargs) except Exception as e: if settings.agent_verbose: - logger.error("TOOL! %s exception=%s", name, e) + logger.error("TOOL! [%s] %s exception=%s", + get_run_tag() or "-", name, e) raise if settings.agent_verbose and t0 is not None: elapsed = time.perf_counter() - t0 - preview = str(result)[:500].replace("\n", " | ") - logger.info("TOOL %s elapsed=%.2fs result=%s", name, elapsed, preview) + preview = str(result)[:VERBOSE_PREVIEW_CHARS].replace("\n", " | ") + logger.info("TOOL [%s] %s elapsed=%.2fs result=%s", + get_run_tag() or "-", name, elapsed, preview) return result return wrapper return decorator @@ -472,20 +509,21 @@ async def update_entity( ) -> str: """Update an entity's mutable fields. Any unspecified field is left unchanged. - IMPORTANT: `content` on a datasource is the original document body and is - read-only via this tool. Any `content` passed for a datasource is dropped - and the tool returns a warning. Use the `notes` field for analysis/summary. + IMPORTANT: `content` is read-only via this tool for a datasource (it is the + original document body — use `notes` for analysis) and for a wiki (its body + belongs to the section tools). Any `content` passed for those is dropped and + the tool returns a warning; every other field still applies. Args: entity_id: UUID of the entity. - content: New content (ignored for datasources). + content: New content (ignored for datasources and wikis). keywords: New keywords list (replaces current). notes: New notes. importance: New importance 0-1. """ try: - # Datasource guardrail — look up type and strip content if protected. - content_dropped = False + # Body guardrail — look up type and strip content if protected. + content_dropped = "" with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT entity_type FROM entities WHERE id = %s", (entity_id,)) @@ -493,9 +531,17 @@ async def update_entity( if not row: return _err(f"entity {entity_id} not found") entity_type = row[0] - if content is not None and entity_type == "datasource": + if content == "": + # An empty string here is destruction, not an edit — observed + # live wiping a thought whose ref was already cited inside a + # wiki body. A real rewrite passes real text; removal is + # delete_entity's job. Ignored with a warning, never silent. content = None - content_dropped = True + content_dropped = ("empty content ignored — use delete_entity " + "to remove an entity, or notes for commentary") + if content is not None and entity_type in _CONTENT_READONLY: + content = None + content_dropped = _CONTENT_READONLY[entity_type] fields = {} if content is not None: @@ -508,7 +554,7 @@ async def update_entity( fields["importance"] = importance if not fields: if content_dropped: - return "No changes (content ignored: datasource bodies are read-only; use notes for analysis)" + return f"No changes (content ignored: {content_dropped})" return "No changes." sets = ", ".join(f"{k} = %s" for k in fields) with get_conn() as conn: @@ -522,7 +568,7 @@ async def update_entity( log_activity(conn, "update", None, entity_id, details={"fields": list(fields.keys())}) msg = f"Updated entity {entity_id}" if content_dropped: - msg += " (content ignored: datasource bodies are read-only; use notes for analysis)" + msg += f" (content ignored: {content_dropped})" return msg except Exception as e: return _err(str(e)) @@ -687,6 +733,22 @@ async def search_sql(query: str) -> str: this entity" — that's view_tree. If you're using SQL to find or understand something, stop and pick the right tool. + Dialect: PostgreSQL (15+). String search: `position()`, `strpos()`, + `regexp_matches`, `regexp_count`, `regexp_instr`, `regexp_substr`, + `regexp_like`, `substring()`. Schema: + entities(id, entity_type, title, content, summary, keywords TEXT[], + importance, notes, metadata, created_at, updated_at, + accessed_at, access_count) + relations(id, from_entity_id, to_entity_id, relation_type, + relevance_score, importance_score, is_bidirectional, + description, notes, created_at, updated_at) + wiki_job(id, job_type, status, target_wiki_id, entity_ids UUID[], + dedupe_key, rationale, proposed_name, batch_id, created_at, + assigned_at, completed_at, attempts, last_error) + wikis_ext(entity_id, canonical_name, disambiguation, language, + member_keyword_ids UUID[], revision, retired_at, redirect_to) + Array columns need `= ANY(col)` or `unnest(col)`, never `IN (col)`. + Args: query: SQL query — must start with SELECT or WITH. """ @@ -835,7 +897,24 @@ async def ingest_file( # DELEGATION — spawn a subagent for focused work # # ====================================================================== # -_call_depth = 0 +# Delegation depth, scoped to the current run context — NOT a module global. +# +# A plain global counts every delegation in flight across the PROCESS. With +# WIKI_WRITE_PARALLELISM writers plus a maintainer sharing one event loop, +# one agent's in-flight delegation made every OTHER agent's next delegation +# fail with "max delegation depth reached" — bounding BREADTH while trying to +# bound DEPTH. Observed live: two sibling delegations in a single parallel +# tool batch 1.8ms apart, the second rejected; 25 such refusals in 70h, +# including a writer's mandatory identity-resolution step, whose rejection +# message then told it to do the work inline. +# +# A ContextVar is inherited by the child Tasks the SDK runs tool bodies in, +# so a subagent spawned from here sees depth+1 and is correctly barred from +# delegating further, while a concurrent sibling run has its own context and +# is unaffected. Unlike run_state's submit slot this needs no mutable +# container: the value is only ever read DOWNWARD into nested runs, never +# written back up to the caller. +_depth_var: ContextVar[int] = ContextVar("braindb_delegation_depth", default=0) _MAX_DEPTH = 1 @@ -845,18 +924,27 @@ async def delegate_to_subagent(task: str) -> str: """Delegate a focused task to a fresh subagent running in its own context. Use for deep searches, duplicate-finding, relation work, or any task where you want the result without polluting your main context with intermediate - tool outputs. The subagent has access to all the same BrainDB tools. + tool outputs. + + The subagent gets the memory tools (including the save/relation tools) + plus the wiki READ tools (`read_wiki_outline`, `read_wiki_section`, + `check_members_cited`, `validate_wiki`). It has NO wiki write tools — + no `edit_wiki_section`, no `delete_wiki_section`, no handoff — so ask + it to investigate, read and report, never to perform a WIKI edit: it + has no safe way to make one. - Write a clear, self-contained task description — the subagent doesn't see - your prior context. End by telling it to call final_answer with a summary. + Its answer comes back to you as ONE string, so ask for a distilled result, + not a transcript. Write a clear, self-contained task description — the + subagent doesn't see your prior context. End by telling it to call + final_answer with a summary. Args: task: A self-contained task description for the subagent. """ - global _call_depth - if _call_depth >= _MAX_DEPTH: + depth = _depth_var.get() + if depth >= _MAX_DEPTH: return "ERROR: max delegation depth reached. Do the task yourself." - _call_depth += 1 + token = _depth_var.set(depth + 1) try: # Local imports to avoid circular dependency on agent.py from braindb.agent.agent import get_subagent, run_typed @@ -878,7 +966,7 @@ async def delegate_to_subagent(task: str) -> str: logger.exception("Subagent failed") return _err(f"subagent failed: {e}") finally: - _call_depth -= 1 + _depth_var.reset(token) # ====================================================================== # @@ -916,15 +1004,20 @@ async def read_wiki_outline(wiki_id: str) -> str: with get_conn() as conn: fetched = ws.fetch_wiki_for_section_op(conn, wiki_id) if fetched is None: - return _err(f"wiki not found: {wiki_id}") + return _wiki_not_found(wiki_id) body, revision = fetched - _, sections = ws.parse_sections(body) + header, sections = ws.parse_sections(body) if not sections: return _err( f"wiki {wiki_id} body has no markers " f"(strict-markers contract violated; cannot edit)" ) - lines = [f"revision: {revision}", f"sections: {len(sections)}"] + lines = [ + f"revision: {revision}", + f'header: {len(header)}ch (meta + title + Summary/Disambiguation' + f' — readable and replaceable as section "header")', + f"sections: {len(sections)}", + ] for s in sections: lines.append(f" - {s.name}: {s.char_count}ch") return "\n".join(lines) @@ -934,27 +1027,93 @@ async def read_wiki_outline(wiki_id: str) -> str: @function_tool @_verbose("read_wiki_section") -async def read_wiki_section(wiki_id: str, section_name: str) -> str: +async def read_wiki_section( + wiki_id: str, + section_name: str, + offset: int = 0, + limit: Optional[int] = None, +) -> str: """Read one section's content + the wiki's current revision token. + A section bigger than one slice is PAGED, never silently cut: the reply + carries `content_meta` {total_chars, offset, returned, next_offset}. Loop + `next_offset` until it is null to hold the whole section — do that before + any `mode="replace"` edit, or you will drop what you never read. + + Read the section you are about to change: where new material belongs, and + how it should be phrased against what is already there, is your judgement. + Args: wiki_id: The wiki's entity UUID. - section_name: Section name as listed by read_wiki_outline. + section_name: Section name as listed by read_wiki_outline, or the + reserved name "header" for the block above the first marker + (meta line, title, Summary/Disambiguation callouts). + offset: start char of the slice (default 0). + limit: max chars of this slice (clamped to the server slice max). """ try: with get_conn() as conn: fetched = ws.fetch_wiki_for_section_op(conn, wiki_id) if fetched is None: - return _err(f"wiki not found: {wiki_id}") + return _wiki_not_found(wiki_id) body, revision = fetched - _, sections = ws.parse_sections(body) + header, sections = ws.parse_sections(body) + if section_name == "header": + chunk, meta = slice_content(header, offset, limit) + return ( + f"revision: {revision}\nsection: header\n" + f"content_meta: {json.dumps(meta)}\ncontent:\n{chunk}" + ) match = next((s for s in sections if s.name == section_name), None) if match is None: names = ", ".join(s.name for s in sections) or "(none)" return _err(f"section '{section_name}' not found. Existing: {names}") - return _truncate( + # Sliced, NOT _truncate'd: the slice is already bounded by SLICE_MAX, + # and a bare truncation here would hide from the model that content + # exists past the cap — which is precisely how a "preserve every prior + # claim" replace turns into silent data loss on a large section. + chunk, meta = slice_content(match.content, offset, limit) + return ( f"revision: {revision}\nsection: {match.name}\n" - f"content:\n{match.content}" + f"content_meta: {json.dumps(meta)}\ncontent:\n{chunk}" + ) + except Exception as e: + return _err(str(e)) + + +@function_tool +@_verbose("check_members_cited") +async def check_members_cited(wiki_id: str, entity_ids: list[str]) -> str: + """Check which of these entity ids are ALREADY cited in the wiki body. + + One cheap call, exact answer — the SAME check the router runs when your + run ends, so it tells you directly whether any citation work is left. It + answers COVERAGE only, not placement: you still read any section you + intend to change. An id reported `gone` no longer exists in the store + and cannot (and need not) be cited. + + Args: + wiki_id: The wiki's entity UUID. + entity_ids: The ids to check — normally the MEMBERS of your job. + """ + try: + with get_conn() as conn: + fetched = ws.fetch_wiki_for_section_op(conn, wiki_id) + if fetched is None: + return _wiki_not_found(wiki_id) + body, revision = fetched + # The shared predicate — identical to the router's end-of-run + # gate, so the two can never disagree. + missing, gone = wj.uncited_members(conn, body, entity_ids) + not_cited = set(missing) | set(gone) + present = [e for e in entity_ids if e not in not_cited] + return ( + f"revision: {revision}\n" + f"cited: {len(present)}/{len(entity_ids)}\n" + f"already_cited: {', '.join(present) or '(none)'}\n" + f"NOT_cited: {', '.join(missing) or '(none)'}\n" + f"gone (deleted from store, cannot be cited): " + f"{', '.join(gone) or '(none)'}" ) except Exception as e: return _err(str(e)) @@ -967,20 +1126,45 @@ async def edit_wiki_section( section_name: str, new_content: str, expect_revision: int, + mode: str = "replace", ) -> str: - """Replace one section's content. If section_name is new, appends a - fresh section at the end. Revision mismatch → returns ERROR: re-read - first. + """Replace a section, or append to it. If section_name is new, a fresh + section is created at the end. Revision mismatch → returns ERROR: + re-read first. + + The reserved name "header" edits the block ABOVE the first section + marker — the `` line, the `# Title`, and the + `> **Summary:**` / `> **Disambiguation:**` callouts. Replace-only: + read it first, keep the meta line, and update the Summary whenever the + page's story has changed — a summary asserting what the body now + disputes is a coherence defect. Args: wiki_id: The wiki's entity UUID. - section_name: Section to replace (or new section to append). - Use lowercase letters, digits, dashes, underscores only. - new_content: Full new content of the section (without the marker - line — the tool re-emits it). + section_name: Section to edit (or new section to create), or + "header". Use lowercase letters, digits, dashes, underscores. + new_content: With mode="replace", the FULL new content of the + section (without the marker line — the tool re-emits it). With + mode="append", ONLY the text to add at the end; existing + content is preserved (trailing blank lines collapse to one). expect_revision: Revision token from the last read on this wiki. + mode: "replace" (default) rewrites the section — read it in full + first, because anything you do not re-emit is gone. "append" + adds at the end, preserving what is there. Choose by content: + integrate/revise when the material relates to existing claims; + append when it is genuinely additive. The header is + replace-only. """ - if not _SECTION_NAME_RE.fullmatch(section_name): + if mode not in ("replace", "append"): + return _err(f"invalid mode '{mode}': use 'replace' or 'append'") + if section_name == "header": + if mode == "append": + return _err( + 'cannot append to "header": it is one block (meta line, ' + 'title, Summary/Disambiguation) — read it, then ' + 'mode="replace" it whole' + ) + elif not _SECTION_NAME_RE.fullmatch(section_name): return _err( f"invalid section_name '{section_name}': use only letters, " f"digits, dashes, underscores" @@ -989,7 +1173,7 @@ async def edit_wiki_section( with get_conn() as conn: fetched = ws.fetch_wiki_for_section_op(conn, wiki_id) if fetched is None: - return _err(f"wiki not found: {wiki_id}") + return _wiki_not_found(wiki_id) body, current_rev = fetched if current_rev != expect_revision: return _err( @@ -1002,16 +1186,24 @@ async def edit_wiki_section( f"wiki {wiki_id} body has no markers; " f"strict-markers contract violated" ) - appended = all(s.name != section_name for s in sections) - new_body = ws.splice_section(body, section_name, new_content) + created = (section_name != "header" + and all(s.name != section_name for s in sections)) + if section_name == "header": + new_body = ws.replace_header(body, new_content) + elif mode == "append": + new_body = ws.append_to_section(body, section_name, new_content) + else: + new_body = ws.splice_section(body, section_name, new_content) new_rev = ws.apply_section_write(conn, wiki_id, new_body, expect_revision) log_activity(conn, "update", "wiki", wiki_id, details={ "op": "edit_wiki_section", "section": section_name, - "appended": appended, + "mode": mode, + "created": created, "revision": new_rev, }) - verb = "appended" if appended else "replaced" + verb = "created" if created else ("appended to" if mode == "append" + else "replaced") return f"ok — section '{section_name}' {verb}. new revision: {new_rev}" except ws.StaleRevisionError as e: return _err(str(e)) @@ -1027,17 +1219,30 @@ async def delete_wiki_section( expect_revision: int, ) -> str: """Remove a section. Revision mismatch → ERROR: re-read first. + The "header" is not deletable — it is the only place the meta line and + the `> **Summary:**` callout can live. Args: wiki_id: The wiki's entity UUID. - section_name: Section to remove. + section_name: Section to remove ("header" is refused). expect_revision: Revision token from the last read on this wiki. """ + if section_name == "header": + return _err( + 'the "header" cannot be deleted — it holds the meta line and ' + 'the Summary/Disambiguation callouts. To change it, use ' + 'edit_wiki_section("header", ..., mode="replace").' + ) + if not _SECTION_NAME_RE.fullmatch(section_name): + return _err( + f"invalid section_name '{section_name}': use only letters, " + f"digits, dashes, underscores" + ) try: with get_conn() as conn: fetched = ws.fetch_wiki_for_section_op(conn, wiki_id) if fetched is None: - return _err(f"wiki not found: {wiki_id}") + return _wiki_not_found(wiki_id) body, current_rev = fetched if current_rev != expect_revision: return _err( @@ -1077,7 +1282,7 @@ async def validate_wiki(wiki_id: str) -> str: with get_conn() as conn: fetched = ws.fetch_wiki_for_section_op(conn, wiki_id) if fetched is None: - return _err(f"wiki not found: {wiki_id}") + return _wiki_not_found(wiki_id) body, revision = fetched issues = ws.check_grammar(body) if not issues: diff --git a/braindb/config.py b/braindb/config.py index f572d0d..ba42b34 100644 --- a/braindb/config.py +++ b/braindb/config.py @@ -34,10 +34,14 @@ "api_key_env": "VLLM_API_KEY", "base_url": "http://host.docker.internal:8002/v1", }, + # Qwen 3.8 27B (NVFP4) on the workstation vLLM. This is the profile the wiki + # pipeline is actually tuned against — the turn budget, request timeout and + # handoff budget below all cite it. Select it and no AGENT_MODEL override is + # needed. "vllm_workstation_qwen": { - "model": "openai/cyankiwi/Qwen3.6-27B-AWQ-INT4", + "model": "openai/QUASAR-QAT/Qwen3.8-27B-QUASAR-NVFP4", "api_key_env": "VLLM_API_KEY", - "base_url": "http://host.docker.internal:8010/v1", + "base_url": "http://host.docker.internal:8012/v1", }, "vllm_workstation_gemma": { "model": "openai/cyankiwi/gemma-4-31B-it-AWQ-4bit", @@ -148,6 +152,20 @@ class Settings(BaseSettings): agent_max_turns: int = 20 agent_subagent_max_turns: int = 30 agent_verbose: bool = False + # Per-LLM-call HTTP deadline (seconds), passed through to LiteLLM as + # `timeout=`. Without it LiteLLM applies its own 600s fallback: its + # `request_timeout` default is the sentinel 6000, and chat `completion()` + # maps that sentinel down to `COMPLETION_HTTP_FALLBACK_SECONDS` (600) + # whenever the caller sets no explicit timeout. 600s is ample for a + # short recall, but a wiki writer regenerating a 50k-char body on a + # local 27B can exceed it — the client then abandons a request vLLM is + # still completing, so the work is computed and discarded. Set to 4800 + # (80 min) to cover a full 30-turn writer run at self-hosted latencies. + # Hosted providers finish far inside this and are unaffected: the value + # is a ceiling, not a delay. NOTE: setting the `REQUEST_TIMEOUT` env var + # to exactly 6000 is a no-op (it IS the sentinel); this setting avoids + # that trap by passing the value per-call instead. + agent_request_timeout: int = 4800 # Runtime "start wrapping up, you have N turns left" nudge (Layer 3 of # Stage C). When ≤ this many LLM-call turns remain before `max_turns` @@ -194,11 +212,41 @@ class Settings(BaseSettings): # initial prompt construction long before the handoff can help. # Default was 9000 during the Phase-3 dry run; observation showed # that fired the handoff on routine consolidates that fit inline - # on Qwen, fragmenting work across successors unnecessarily. Set - # to 0 to disable the handoff nudge entirely. - agent_writer_handoff_token_budget: int = 20000 + # on Qwen, fragmenting work across successors unnecessarily. + # Raised 20000 -> 30000 after a long soak: on pages approaching + # 50k chars the writer needs the successor path, and a budget set + # too high simply never fires, leaving it to accumulate context + # until it hits its turn limit. Keep this well inside the smallest + # context window you deploy on — above it, handoff is silently + # dead. Set to 0 to disable the handoff nudge entirely. + agent_writer_handoff_token_budget: int = 30000 agent_writer_handoff_max_depth: int = 3 + # Reasoning effort for the WIKI agents (maintainer / writer / subagent). + # Blank = send nothing, i.e. the server-side default — no behaviour + # change. Delivered as `chat_template_kwargs.reasoning_effort` in the + # request body (see `agent._build` for why it cannot be sent as a plain + # `reasoning_effort` param). SELF-HOSTED vLLM ONLY: a hosted provider may + # reject the unknown body key, so leave it blank on those profiles. + # + # Why this exists: the Qwen3 chat template resolves + # `reasoning_effort|default('xhigh')`, so sending nothing runs every + # request at the MAXIMUM setting and injects a "think carefully, + # validate assumptions, consider alternatives" instruction into each + # system block. Measured on the bench box: ~2750 output tokens/turn + # and ~99% of wall clock is generation, so trimming reasoning trims + # latency almost linearly. It costs no cross-turn consistency either: + # the SDK only replays reasoning for DeepSeek/Claude models, so on + # Qwen every reasoning token is generated, paid for and then dropped + # before the next turn. + # + # Valid values for that template: "low", "medium", "none". NOT + # "minimal"/"high" — vLLM's Literal accepts them but the template's + # own validator raises. Wiki-scoped on purpose: the general agent + # (`get_agent`) is shared with the ingest watcher, whose extraction + # runs are the most reasoning-dependent work in the stack. + agent_wiki_reasoning_effort: str = "" + @property def resolved_agent_model(self) -> str: return self.agent_model or _LLM_PROFILES[self.llm_profile]["model"] diff --git a/braindb/main.py b/braindb/main.py index 21b4289..b1875c6 100644 --- a/braindb/main.py +++ b/braindb/main.py @@ -1,17 +1,25 @@ import logging +from datetime import datetime, timezone from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware +from braindb.db import get_conn from braindb.routers import agent, entities, integrations, memory, relations, wiki +from braindb.services import wiki_jobs +from braindb.services.activity_log import log_activity from braindb.services.embedding_service import get_embedding_service logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") +# Captured at import = process start. Startup uses it to release wiki jobs +# whose claims died with the PREVIOUS process (see release_stale_assigned). +_PROCESS_START = datetime.now(timezone.utc) + app = FastAPI( title="BrainDB", description="Memory database and REST API for LLM agents", - version="0.9.0", + version="0.10.0", ) app.add_middleware( @@ -33,7 +41,28 @@ @app.on_event("startup") def startup(): - """Initialize the embedding service on startup.""" + """Release restart-orphaned wiki claims, then load embeddings.""" + # Agent runs execute only inside this process (scheduler/watcher are + # HTTP clients), so any wiki_job still `assigned` from before this + # process started belongs to a run that no longer exists. Release them + # now instead of letting them sit dark for the full lease. Runs before + # the slow embedding load so recovery is immediate; the schema is + # guaranteed present because migrations run in the container command + # before uvicorn (see docker-compose*.yml). + try: + with get_conn() as conn: + released = wiki_jobs.release_stale_assigned(conn, _PROCESS_START) + if released: + log_activity(conn, "wiki_jobs_release", None, None, + details={"released": released, + "reason": "assigned before process start"}) + logging.getLogger(__name__).info( + "released %d restart-orphaned wiki job claim(s)", released) + except Exception: + # A DB hiccup here must never block the API from starting — the + # lease remains the fallback for anything not released. + logging.getLogger(__name__).exception( + "startup release of stale assigned jobs failed; lease will cover") emb = get_embedding_service() emb.initialize() diff --git a/braindb/routers/wiki.py b/braindb/routers/wiki.py index 97e88eb..fcb03f0 100644 --- a/braindb/routers/wiki.py +++ b/braindb/routers/wiki.py @@ -118,7 +118,10 @@ async def wiki_maintain(): # EACH seed's own subject (recall_memory / view_tree / delegate) before # deciding — co-occurrence is not identity. Returns one decision per seed. catalog_txt = ( - "\n".join(f"{i}. {w['canonical_name']}" for i, w in enumerate(cat, 1)) + "\n".join( + f"{i}. {w['canonical_name']} ({w['char_count']}ch)" + for i, w in enumerate(cat, 1) + ) or "(no existing wikis yet — attach/consolidate are impossible; " "use create/skip/ambiguous)" ) @@ -141,6 +144,13 @@ async def wiki_maintain(): with get_conn() as conn: wiki_jobs.finish_jobs(conn, [s["job_id"] for s in seeds], "failed", f"agent error: {e}"[:500]) + # A crashed run must be visible in the DB audit trail too, not + # only in wiki_job.last_error and the container log. + log_activity(conn, "wiki_maintain", None, None, details={ + "result": "failed", + "jobs": [s["job_id"] for s in seeds], + "error": str(e)[:500], + }) return {"claimed": len(jobs), "result": "failed", "reason": str(e)} by_id = {d.entity_id: d for d in res.decisions} @@ -227,7 +237,8 @@ def _apply_decision(conn, *, decision: dict, orphan: dict, job_id: str, wiki_jobs.finish_job(conn, job_id, "failed", "create missing proposed_name") outcome = {"action": "create", "error": "missing proposed_name"} else: - key = wiki_jobs.suggestion_dedupe_key("create", None, [orphan_id], []) + key = wiki_jobs.suggestion_dedupe_key( + "create", None, [orphan_id], [], proposed_name=name) sid = wiki_jobs.insert_suggestion( conn, job_type="create", target_wiki_id=None, entity_ids=[orphan_id], dedupe_key=key, rationale=rationale, @@ -285,6 +296,12 @@ def _members_block(members: list[dict]) -> str: # exactly for navigating a body without inlining it. _INLINE_BODY_MAX_CHARS = 4000 +# Above this character count the page is already long enough that adding more +# prose costs more than it informs (~10k tokens). The stub gains one paragraph +# telling the writer to stay dense and name a neighbouring page rather than +# expand this one. Advisory only — no gate, no rejection. +_DENSITY_NUDGE_CHARS = 30000 + def _body_block_or_stub(mode: str, wiki_id: str | None, old_body: str) -> str: """For attach mode with a body too large to safely inline, return a @@ -293,6 +310,18 @@ def _body_block_or_stub(mode: str, wiki_id: str | None, old_body: str) -> str: if not old_body: return "(none — create mode)" if mode == "attach" and wiki_id and len(old_body) > _INLINE_BODY_MAX_CHARS: + nudge = "" + if len(old_body) > _DENSITY_NUDGE_CHARS: + nudge = ( + f"\n\nThis page is ALREADY LONG ({len(old_body)} chars). Prefer\n" + f"density over growth: integrate into existing prose rather than\n" + f"append parallel sentences, and where a NON-member detail you\n" + f"turned up belongs to a neighbouring subject, name that page in\n" + f"prose instead of restating it here — or leave it out; it stays\n" + f"an orphan and comes back for its own page later. This never\n" + f"applies to a MEMBER of this job: every member must be cited\n" + f"here regardless of length." + ) return ( f"[BODY OMITTED — {len(old_body)} chars, too large to inline.\n" f"Use the section tools to navigate without consuming context:\n" @@ -301,6 +330,7 @@ def _body_block_or_stub(mode: str, wiki_id: str | None, old_body: str) -> str: f" - edit_wiki_section(...) per section, validate_wiki, then\n" f" final_answer(mode=\"attach\", body=\"\") — router persists via\n" f" section edits and skips the full-body write.]" + f"{nudge}" ) return old_body @@ -342,6 +372,18 @@ async def wiki_write(): return {"written": 0, "result": "failed", "reason": "target wiki missing"} canonical = wiki["canonical_name"] old_body = wiki["content"] or "" + # Snapshot at CLAIM, not at persist. The snapshot's CONTENT was + # always this claim-time body — only its WRITE used to happen at + # persist, which meant a run that died after a section edit left + # a committed mutation with no snapshot anywhere. Writing it here + # (same transaction as the claim) closes that window; the section + # tools' own commits are now always preceded by a durable + # "this run started from revision N" record. A run that later + # fails or no-ops leaves this row too — truthful, and bounded by + # max_attempts. + wiki_jobs.snapshot_revision( + conn, str(bucket["target_wiki_id"]), old_body, + wiki_jobs.parse_refs(old_body), wiki["revision"]) elif mode == "consolidate": members = [] dupes = wiki_jobs.fetch_wikis_for_merge(conn, bucket["wiki_ids"]) @@ -352,12 +394,39 @@ async def wiki_write(): canonical = "(decide among duplicates)" wiki = None old_body = "\n\n".join(d["content"] or "" for d in dupes) + # Same claim-time snapshot rule as attach, one per duplicate + # (this now precedes the canonical_no validation at persist — + # a snapshot of a merge that then failed validation is a + # harmless, truthful record). + for d in dupes: + wiki_jobs.snapshot_revision( + conn, d["id"], d["content"] or "", + wiki_jobs.parse_refs(d["content"] or ""), d["revision"]) else: # create members = wiki_jobs.fetch_members(conn, member_ids) canonical = bucket["proposed_name"] or "Untitled" wiki = None old_body = "" batch_id = str(jobs[0].get("batch_id")) if jobs[0].get("batch_id") else None + # Neighbouring pages the writer may name in prose instead of expanding + # this one. Short and relevant by construction — NOT the full catalog. + # Never in consolidate mode: there `member_ids` holds the DUPLICATE + # WIKI ids, so this would hand the writer the very pages it must + # absorb and invite it to link out to them instead of merging them. + related = ( + [] + if mode == "consolidate" + else wiki_jobs.list_related_wikis( + conn, member_ids, bucket.get("target_wiki_id") + ) + ) + + def _related_wikis_block(rows: list[dict]) -> str: + if not rows: + return "(none — no neighbouring pages exist yet)" + return "\n".join( + f"- {r['canonical_name']} ({r['char_count']}ch)" for r in rows + ) def _dupes_block(ds: list[dict]) -> str: if not ds: @@ -383,6 +452,7 @@ def _dupes_block(ds: list[dict]) -> str: .replace("%%WIKI_ID%%", bucket["target_wiki_id"] or "(assigned after write)") .replace("%%MEMBERS%%", _members_block(members)) .replace("%%CURRENT_BODY%%", _body_block_or_stub(mode, bucket.get("target_wiki_id"), old_body)) + .replace("%%RELATED_WIKIS%%", _related_wikis_block(related)) .replace("%%DUPLICATES%%", _dupes_block(dupes)) ) + _now_line() + custom_profile.additions("wiki_writer") # Capture pre-run revision on the target wiki for `attach` mode so we @@ -430,21 +500,36 @@ def _dupes_block(ds: list[dict]) -> str: max_depth = settings.agent_writer_handoff_max_depth while handoff_slot.captured and depth < max_depth: depth += 1 + # Appended AFTER the rendered job prompt (see run_typed call + # below), so the successor really does have the same prompt: + # mode, canonical name, wiki id, MEMBERS, the body stub, the + # neighbouring pages and every writer rule. Seeding it with + # this brief ALONE — as this loop used to — left it running on + # the generic SYSTEM_PROMPT with no members, no preservation + # rule and no `body=""` contract, so it re-read from scratch + # exactly what the brief had just told it. seed = ( - "Continuing from a previous agent run that ended early " - "via `handoff_to_successor` because its context was " - "filling up. You have the SAME prompt, the SAME tools, " - "and a fresh context window. Resume from this state.\n\n" + "---\n\n" + "CONTINUATION. The job above is unchanged. A previous " + "agent run on it ended early via `handoff_to_successor` " + "because its context was filling up. You have its brief " + "below, the same tools, and a fresh context window.\n\n" "PROGRESS SO FAR (from the previous agent):\n" f"{handoff_slot.progress_summary}\n\n" "REMAINING WORK:\n" f"{handoff_slot.remaining_work}\n\n" - "Pick up from here. Call `final_answer` when done " - "(body=\"\" if you persisted via section-edit tools, " - "or the full body otherwise). If YOUR context also " - "fills up before you finish, call `handoff_to_successor` " - "again with an updated brief — the same successor " - "mechanism will continue." + "Trust the brief for STATE — revisions, decisions made, " + "sections already done — and do not re-derive those. But " + "your fresh context does not hold any section's TEXT: " + "still read any section (or the header) you intend to " + "replace, and choose edits by content, not by cost, " + "exactly as the job instructions above say. Call " + "`final_answer` when done — body=\"\" is for ATTACH mode " + "only (when your section edits persisted the content); " + "create and consolidate must submit the full body. If " + "YOUR context also fills up before you finish, call " + "`handoff_to_successor` again with an updated brief — " + "the same successor mechanism will continue." ) handoff_slot.captured = False handoff_slot.progress_summary = "" @@ -454,7 +539,8 @@ def _dupes_block(ds: list[dict]) -> str: depth, max_depth, mode, job_ids, ) res = await run_typed( - seed, get_writer_agent(), WikiWriteResult, max_turns=30, + f"{prompt}\n\n{seed}", get_writer_agent(), WikiWriteResult, + max_turns=30, token_budget=settings.agent_writer_handoff_token_budget, ) if handoff_slot.captured: @@ -470,17 +556,29 @@ def _dupes_block(ds: list[dict]) -> str: f"handoff depth cap {max_depth} exhausted " f"without final_answer", ) + log_activity(conn, "wiki_write", "wiki", + bucket.get("target_wiki_id"), details={ + "result": disp, "mode": mode, + "jobs": job_ids, + "error": "handoff depth exhausted", + }) return {"written": 0, "result": disp, "reason": "handoff depth exhausted"} except Exception as e: logger.exception("writer agent failed") with get_conn() as conn: disp = wiki_jobs.release_or_fail_jobs(conn, job_ids, f"agent error: {e}") + # Mirror the success-path wiki_write row so a DB-only + # auditor sees crashed runs, not a suspiciously clean log. + log_activity(conn, "wiki_write", "wiki", + bucket.get("target_wiki_id"), details={ + "result": disp, "mode": mode, + "jobs": job_ids, "error": str(e)[:500], + }) return {"written": 0, "result": disp, "reason": str(e)} finally: release_handoff_slot(handoff_token) - used_section_edits = False if _is_blank_body(res.body): # Empty body — only valid in attach mode if section edits bumped # the revision during the run. Otherwise the agent did nothing @@ -516,19 +614,24 @@ def _dupes_block(ds: list[dict]) -> str: # `summarises` relations catch up. If any member is missing # from the body, the writer skipped real work — fail it. body_now = row[0] or "" - cited = wiki_jobs.parse_refs(body_now) # lower-cased set - missing = [m for m in member_ids if m.lower() not in cited] - if missing: - with get_conn() as conn: + with get_conn() as conn: + # The shared predicate (`uncited_members`) splits un-cited + # members into `missing` (entity exists — real outstanding + # work) and `gone` (entity deleted since triage — can never + # be cited, and must not wedge the job in a retry loop; the + # same premise as reconcile's dangling-ref skip). + missing, gone = wiki_jobs.uncited_members( + conn, body_now, member_ids) + if missing: disp = wiki_jobs.release_or_fail_jobs( conn, job_ids, f"empty body AND no section edits AND " f"{len(missing)} member(s) not yet cited in body", ) - return {"written": 0, "result": disp, - "reason": "members un-cited"} - # All members cited — close the no-op cleanly and reconcile. - with get_conn() as conn: + return {"written": 0, "result": disp, + "reason": "members un-cited"} + # Every existing member cited — close the no-op cleanly + # and reconcile. `gone` ids are recorded, never silent. rel = wiki_jobs.reconcile_summarises_additive( conn, bucket["target_wiki_id"], body_now) wiki_jobs.finish_jobs(conn, job_ids, "done") @@ -536,18 +639,18 @@ def _dupes_block(ds: list[dict]) -> str: bucket["target_wiki_id"], details={ "mode": mode, "no_op": True, "revision": pre_revision, - "members": len(member_ids), **rel, + "members": len(member_ids), + "members_gone": gone, **rel, }) logger.info( "writer no-op accepted: pre_rev=%s, all %d members already " - "cited; reconcile=%s", - pre_revision, len(member_ids), rel, + "cited (%d gone); reconcile=%s", + pre_revision, len(member_ids), len(gone), rel, ) return {"written": 0, "wiki_id": bucket["target_wiki_id"], "mode": mode, "revision": pre_revision, "jobs": job_ids, "no_op": True, **rel} new_body = row[0] - used_section_edits = True logger.info( "writer used section-edit path: pre_rev=%s post_rev=%s body=%dch", pre_revision, row[1], len(new_body), @@ -593,10 +696,8 @@ def _dupes_block(ds: list[dict]) -> str: "reason": "invalid canonical_no"} canonical_id = dupes[no - 1]["id"] wiki_id = canonical_id - for d in dupes: - wiki_jobs.snapshot_revision( - conn, d["id"], d["content"] or "", - wiki_jobs.parse_refs(d["content"] or ""), d["revision"]) + # Snapshots were written at CLAIM time (see the claim block) — + # every duplicate's pre-run body is already durably recorded. revision = wiki_jobs.finalize_wiki_write( conn, wiki_id, new_body, summary, disambig, member_ids) for d in dupes: @@ -605,9 +706,7 @@ def _dupes_block(ds: list[dict]) -> str: retired.append(d["id"]) else: # attach wiki_id = bucket["target_wiki_id"] - wiki_jobs.snapshot_revision( - conn, wiki_id, old_body, wiki_jobs.parse_refs(old_body), - wiki["revision"]) + # Snapshot was written at CLAIM time (see the claim block). revision = wiki_jobs.finalize_wiki_write( conn, wiki_id, new_body, summary, disambig, member_ids) diff --git a/braindb/services/wiki_jobs.py b/braindb/services/wiki_jobs.py index 5a71be4..71a94f9 100644 --- a/braindb/services/wiki_jobs.py +++ b/braindb/services/wiki_jobs.py @@ -33,10 +33,31 @@ # only while a worker is actively running it; if that worker never returns # (api restart mid-run, agent timeout) the row would wedge forever. Instead # of a reaper/cycle, an `assigned` job whose lease expired is simply -# claimable again at the EXISTING claim step. 20 min is comfortably above -# the longest legit run (AGENT_TIMEOUT ~10 min), so a still-running job is -# never reclaimed. `attempts`+max_attempts already bound repeated failures. -ASSIGNED_LEASE_MIN = int(os.getenv("WIKI_ASSIGNED_LEASE_MIN", "20")) +# claimable again at the EXISTING claim step. +# Raised 20 -> 120. The old value cited "the longest legit run (AGENT_TIMEOUT +# ~10 min)", but that premise expired three bumps ago. OPERATOR INVARIANT: +# the lease must exceed YOUR deployment's longest legitimate write — at +# minimum the scheduler's WIKI_AGENT_TIMEOUT and the per-call +# agent_request_timeout — or a still-running job gets reclaimed and run +# twice. The default 120 min covers the shipped defaults (2400s scheduler +# patience, 4800s per-call ceiling); a deployment that raises those (the +# bench overlay runs WIKI_AGENT_TIMEOUT=18000s = 300 min) must raise +# WIKI_ASSIGNED_LEASE_MIN above them too. Read in the API process (this +# module is imported by the router), so set it on the api service. +ASSIGNED_LEASE_MIN = int(os.getenv("WIKI_ASSIGNED_LEASE_MIN", "120")) + +# Hard ceiling on how many times ONE job may be reclaimed after a lease +# expiry. `release_or_fail_jobs` caps the GRACEFUL failure path at +# max_attempts, but an ABANDONED run (client timeout, 500, worker death) +# never reaches it: `_claimable()` re-admits the row and `claim_jobs` +# increments `attempts` again, with nothing to stop the cycle. Observed live: +# a perfect staircase, one job at every attempts value 1..30, and 24 jobs +# with up to 28 attempts and `last_error` NULL — 832 claim cycles on a single +# wiki whose work was already complete. Past this ceiling the row simply +# stops being claimable: it stays `assigned` and visible on `GET /jobs` for +# an operator, instead of consuming the queue forever. Set above the +# graceful max_attempts (3) so normal failure handling still runs first. +ASSIGNED_MAX_RECLAIMS = int(os.getenv("WIKI_ASSIGNED_MAX_RECLAIMS", "5")) # Per-wiki attach grouping — how long to wait before firing a writer on a # wiki that just received new attaches. Once the OLDEST pending attach for @@ -70,11 +91,16 @@ def _claimable(alias: str = "") -> str: - """SQL predicate: a job is claimable if pending, OR assigned but its - lease expired. Reused verbatim at every claim site (DRY). `alias` is the - table alias when the query qualifies columns (e.g. 'j').""" + """SQL predicate: a job is claimable if pending, OR assigned with its + lease expired AND still under `ASSIGNED_MAX_RECLAIMS`. Reused verbatim at + every claim site (DRY). `alias` is the table alias when the query + qualifies columns (e.g. 'j'). + + A `pending` job is always claimable — the ceiling applies only to the + reclaim branch, which is the one nothing else bounds.""" p = f"{alias}." if alias else "" return (f"({p}status = 'pending' OR ({p}status = 'assigned' " + f"AND {p}attempts < {ASSIGNED_MAX_RECLAIMS} " f"AND {p}assigned_at < now() - make_interval(mins => {ASSIGNED_LEASE_MIN})))") # Inline reference token: [[ref:UUID]] or [[ref:UUID|display text]] @@ -123,9 +149,19 @@ def reconcile_summarises_additive(conn, wiki_id: str, body: str) -> dict: never deletes or re-types a relation behind the LLM. If the LLM wants a relation gone it calls `delete_relation` itself. Mirrors LLM-authored content into the graph; it does not judge or shape content. + + A cited id with no row in `entities` (deleted since it was cited, or a + UUID mistyped while re-emitting a body) is SKIPPED, not inserted: the + FK violation would abort the caller's whole transaction, taking + `finish_jobs` with it — so the jobs re-queue forever while the section + edits already committed during the agent run remain. Bookkeeping must + never be blocked by content. Skipped ids are returned (and land in the + `wiki_write` activity log via the caller's `**rel` spread) so this is + visible, never silent. """ cited = parse_refs(body) added = 0 + skipped: list[str] = [] with conn.cursor() as cur: cur.execute( "SELECT to_entity_id::text FROM relations " @@ -133,7 +169,16 @@ def reconcile_summarises_additive(conn, wiki_id: str, body: str) -> dict: (wiki_id,), ) current = {r[0].lower() for r in cur.fetchall()} - for e in cited - current: + wanted = sorted(cited - current) + if wanted: + cur.execute( + "SELECT id::text FROM entities WHERE id = ANY(%s::uuid[])", + (wanted,), + ) + live = {r[0].lower() for r in cur.fetchall()} + skipped = [e for e in wanted if e not in live] + wanted = [e for e in wanted if e in live] + for e in wanted: cur.execute( """INSERT INTO relations (from_entity_id, to_entity_id, relation_type, relevance_score, description) @@ -142,7 +187,8 @@ def reconcile_summarises_additive(conn, wiki_id: str, body: str) -> dict: (wiki_id, e), ) added += 1 - return {"relations_added": added, "relations_removed": 0} + return {"relations_added": added, "relations_removed": 0, + "refs_skipped": skipped} def try_wiki_lock(conn, key: str) -> bool: @@ -152,6 +198,32 @@ def try_wiki_lock(conn, key: str) -> bool: return bool(cur.fetchone()[0]) +def release_stale_assigned(conn, before) -> int: + """Return to `pending` every job still `assigned` from BEFORE `before` + (normally the API process's start time — see main.py startup). + + Agent runs execute ONLY inside the API process; the scheduler and the + watcher are HTTP clients with no braindb imports. So a row assigned + before this process existed belongs to a run that died with the previous + process — without this it would sit dark for the full lease (observed: + a restart-orphaned consolidate invisible for hours while writers + re-derived the identity it would have settled). The lease remains the + safety net for deaths the process CANNOT see; this handles the one kind + it can. NOTE: keys off PROCESS start — the invariant breaks the day the + api runs with multiple workers or replicas. + + `attempts` is preserved: the next claim increments it as usual, and the + reclaim ceiling + run_cron disposition still bound repeated wedging. + """ + with conn.cursor() as cur: + cur.execute( + """UPDATE wiki_job SET status = 'pending', assigned_at = NULL + WHERE status = 'assigned' AND assigned_at < %s""", + (before,), + ) + return cur.rowcount + + def claim_jobs(conn, job_ids: list[str]) -> int: """Mark a bucket's pending suggestion jobs as assigned (SKIP LOCKED).""" if not job_ids: @@ -241,6 +313,27 @@ def run_cron(conn) -> dict: """ batch_id = str(uuid.uuid4()) with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + # Disposition first: an `assigned` job past both the lease AND the + # reclaim ceiling is un-claimable (`_claimable()` excludes it) and + # nothing else ever closes it — it would wedge forever, and because + # `assigned` counts as an ACTIVE job in `_orphan_conditions()`, its + # entity would be silently barred from re-triage for good. Flipping + # it to `failed` is the status that, by design, returns the entity + # to the orphan pool (`failed` is deliberately not excluded there), + # so the very scan below can re-enqueue it in this same tick. + # Bounded per JOB by the ceiling, self-healing per ENTITY via full + # cron cycles — the original contract, minus the reclaim spin. + cur.execute( + f"""UPDATE wiki_job + SET status = 'failed', + last_error = 'reclaim ceiling ({ASSIGNED_MAX_RECLAIMS}) ' + 'reached; entity returns to the orphan pool' + WHERE status = 'assigned' + AND attempts >= {ASSIGNED_MAX_RECLAIMS} + AND assigned_at < now() - make_interval(mins => {ASSIGNED_LEASE_MIN})""" + ) + assigned_expired_failed = cur.rowcount + cur.execute( f""" WITH orphans AS ( @@ -270,6 +363,7 @@ def run_cron(conn) -> dict: "batch_id": batch_id, "triage_jobs_enqueued": enqueued, "pending_triage_total": pending_triage, + "assigned_expired_failed": assigned_expired_failed, } @@ -394,12 +488,24 @@ def fetch_entity_brief(conn, entity_id: str) -> dict | None: def suggestion_dedupe_key(action: str, target_wiki_id: str | None, - entity_ids: list[str], consolidate_wiki_ids: list[str]) -> str: - """Deterministic, service-computed (never LLM-computed) idempotency key.""" + entity_ids: list[str], consolidate_wiki_ids: list[str], + proposed_name: str | None = None) -> str: + """Deterministic, service-computed (never LLM-computed) idempotency key. + + `create` keys on the PROPOSED NAME, not the seed ids: two maintainer runs + proposing the same page name seconds apart used to mint two pages plus a + consolidate to undo it (observed three runs in a row, 65s apart). With + the name key the second insert conflicts; its orphan re-enters the pool + on the next cron and attaches to the page the first run built. The key + only spans ACTIVE jobs (partial index on pending/assigned), so a later + create for the same name — after the first completed — still inserts. + NOTE: unrelated to the advisory-lock string `create:{job_id}` in the + router (`lock_key`); that is a lock name, not a dedupe key. + """ if action == "attach": return f"attach:{target_wiki_id}:" + ",".join(sorted(entity_ids)) if action == "create": - return "create:" + ",".join(sorted(entity_ids)) + return "create:" + (proposed_name or ",".join(sorted(entity_ids))).lower() if action == "consolidate": return "consolidate:" + ",".join(sorted(consolidate_wiki_ids)) raise ValueError(f"unknown action {action!r}") @@ -499,6 +605,36 @@ def next_write_bucket(conn) -> dict | None: "target_wiki_id": str(seed["target_wiki_id"]), "proposed_name": None} +def uncited_members(conn, body: str, + member_ids: list[str]) -> tuple[list[str], list[str]]: + """(missing, gone): member ids not cited in `body`, split into those + whose entity still exists (real outstanding work) and those with no + entity row. + + A member deleted since triage can never be cited, so treating it like + ordinary "missing" wedges its job in a fail/retry loop forever — the + same premise as `reconcile_summarises_additive`'s dangling-ref skip: + absent content must never block bookkeeping. + + This is THE citation predicate. The router's no-op gate and the + writer's `check_members_cited` tool both call it, so they can never + drift apart. + """ + cited = parse_refs(body) + uncited = [m for m in member_ids if m.lower() not in cited] + if not uncited: + return [], [] + with conn.cursor() as cur: + cur.execute( + "SELECT id::text FROM entities WHERE id = ANY(%s::uuid[])", + (uncited,), + ) + live = {r[0].lower() for r in cur.fetchall()} + missing = [m for m in uncited if m.lower() in live] + gone = [m for m in uncited if m.lower() not in live] + return missing, gone + + def fetch_members(conn, entity_ids: list[str]) -> list[dict]: if not entity_ids: return [] @@ -525,13 +661,17 @@ def fetch_wiki(conn, wiki_id: str) -> dict | None: def list_active_wikis(conn) -> list[dict]: - """All non-retired wikis as {id, canonical_name}, deterministically - ordered. Plumbing read (mirrors fetch_wiki / export_wikis SQL) — the - maintainer is shown this as a NUMBERED catalog so it references wikis by - number, never by uuid; the order here IS the numbering.""" + """All non-retired wikis as {id, canonical_name, char_count}, + deterministically ordered. Plumbing read (mirrors fetch_wiki / + export_wikis SQL) — the maintainer is shown this as a NUMBERED catalog so + it references wikis by number, never by uuid; the order here IS the + numbering. `char_count` lets the maintainer see that a candidate target is + already large and prefer a narrower `create` over piling on another + `attach`; without it every page looks equally empty.""" with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute( - """SELECT e.id::text AS id, w.canonical_name + """SELECT e.id::text AS id, w.canonical_name, + length(e.content) AS char_count FROM entities e JOIN wikis_ext w ON w.entity_id = e.id WHERE e.entity_type = 'wiki' AND w.retired_at IS NULL ORDER BY e.importance DESC, e.created_at""" @@ -539,6 +679,43 @@ def list_active_wikis(conn) -> list[dict]: return [dict(r) for r in cur.fetchall()] +def list_related_wikis(conn, entity_ids: list[str], exclude_wiki_id: str | None, + limit: int = 10) -> list[dict]: + """Wikis one hop from the entities being written, as + {canonical_name, char_count}, most-cited first. + + The WRITER gets this — deliberately NOT the full catalog. It only needs to + know which neighbouring pages already exist so it can name one in prose + instead of expanding this page. A short, relevant list keeps that cheap; + the whole catalog would be noise and grows without bound. + + "Related" = a wiki that already `summarises` an entity that one of these + members is connected to. Reuses the relations the pipeline already + maintains; creates nothing.""" + if not entity_ids: + return [] + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + """SELECT w.canonical_name, + length(e.content) AS char_count, count(*) AS cites + FROM relations seed + JOIN relations sm ON sm.to_entity_id IN (seed.from_entity_id, + seed.to_entity_id) + AND sm.relation_type = 'summarises' + JOIN entities e ON e.id = sm.from_entity_id + JOIN wikis_ext w ON w.entity_id = e.id + WHERE (seed.from_entity_id = ANY(%s::uuid[]) + OR seed.to_entity_id = ANY(%s::uuid[])) + AND e.entity_type = 'wiki' AND w.retired_at IS NULL + AND (%s::uuid IS NULL OR e.id <> %s::uuid) + GROUP BY e.id, w.canonical_name + ORDER BY cites DESC, w.canonical_name + LIMIT %s""", + (entity_ids, entity_ids, exclude_wiki_id, exclude_wiki_id, limit), + ) + return [dict(r) for r in cur.fetchall()] + + def release_or_fail_jobs(conn, job_ids: list[str], last_error: str, max_attempts: int = 3) -> str: """On a gate failure: return jobs to 'pending' for retry, or 'failed' once @@ -626,8 +803,9 @@ def _keyword_ids_among(conn, entity_ids: list[str]) -> list[str]: def finalize_wiki_write(conn, wiki_id: str, new_body: str, summary: str | None, disambiguation: str | None, member_entity_ids: list[str]) -> int: - """Apply the gated body to an existing wiki: update content + header - fields, union new keyword members, bump revision.""" + """Apply the LLM-authored body to an existing wiki: update content + + header fields, union new keyword members, bump revision. (There is no + content gate — the deliberate design; see routers/wiki.py.)""" new_kw = _keyword_ids_among(conn, member_entity_ids) with conn.cursor() as cur: cur.execute("UPDATE entities SET content=%s, summary=%s WHERE id=%s", diff --git a/braindb/services/wiki_sections.py b/braindb/services/wiki_sections.py index 9b8f69d..078f83a 100644 --- a/braindb/services/wiki_sections.py +++ b/braindb/services/wiki_sections.py @@ -19,7 +19,8 @@ silently stomping on a concurrent edit. Pure parsing functions (`parse_sections`, `splice_section`, -`delete_section`, `check_grammar`) are DB-free and unit-testable. +`append_to_section`, `replace_header`, `delete_section`, `check_grammar`) +are DB-free and unit-testable. The two DB helpers at the bottom (`fetch_wiki_for_section_op`, `apply_section_write`) are the only stateful surface. """ @@ -115,6 +116,46 @@ def splice_section(body: str, section_name: str, new_content: str) -> str: return _rebuild(header, sections) +def append_to_section(body: str, section_name: str, added_content: str) -> str: + """Append text to the END of one named section, preserving its existing + content (whitespace-normalised: trailing blank lines collapse to one, + and the rebuild re-emits section markers in canonical form — no claim + or citation can be lost, but the guarantee is content-level, not + byte-level). + + The genuinely-additive case: a new `[[ref:UUID]]` bullet or a new claim. + `splice_section` requires the caller to supply the section's WHOLE new + content; resolving an append here, from the body the caller already + holds, means the model never has to carry the prior content just to add + to it. When new material relates to what a section already says, a + replace that integrates it is the better edit — that judgement belongs + to the caller, not this function. + + A section that doesn't exist yet is created with `added_content`, matching + `splice_section`'s append-a-new-section behaviour. + """ + _, sections = parse_sections(body) + existing = next((s for s in sections if s.name == section_name), None) + if existing is None: + return splice_section(body, section_name, added_content) + merged = existing.content.rstrip("\n") + "\n" + added_content.lstrip("\n") + return splice_section(body, section_name, merged) + + +def replace_header(body: str, new_header: str) -> str: + """Replace the HEADER — everything above the first section marker: the + `` line, the `# Title`, and the `> **Summary:**` / + `> **Disambiguation:**` callouts. Every section is preserved untouched. + + This is the fix for header freeze: section tools could edit every + section but never the block readers see first, so a page's summary + could permanently contradict its own body. Exposed to the writer as + the reserved section name "header" (replace-only). + """ + _, sections = parse_sections(body) + return _rebuild(new_header, sections) + + def delete_section(body: str, section_name: str) -> str: """Remove the named section (and its marker) from the body. Raises KeyError if the section isn't present.""" @@ -157,6 +198,10 @@ def check_grammar(body: str) -> list[str]: issues.append(f"malformed [[ref: token at char offset {m.start()}") if "> **Summary:**" not in header: issues.append("missing > **Summary:** callout in header") + if " line in header") return issues diff --git a/docker-compose.yml b/docker-compose.yml index 2393885..395acd4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -29,6 +29,13 @@ services: DEEPINFRA_API_KEY: ${DEEPINFRA_API_KEY:-} VLLM_API_KEY: ${VLLM_API_KEY:-} AGENT_VERBOSE: ${AGENT_VERBOSE:-false} + # Per-LLM-call HTTP deadline. Default 4800s (80 min) — a ceiling, not a + # delay; without it LiteLLM falls back to 600s and abandons long + # self-hosted writes the server is still completing. + AGENT_REQUEST_TIMEOUT: ${AGENT_REQUEST_TIMEOUT:-4800} + # Reasoning effort for the WIKI agents only. Blank = server default. + # Self-hosted vLLM only; leave blank on hosted providers. + AGENT_WIKI_REASONING_EFFORT: ${AGENT_WIKI_REASONING_EFFORT:-} # Orphan freshness gate (the orphan SQL runs in this api process, not in # the scheduler): an entity is wiki-eligible only once created_at is # older than this many minutes, so still-ingesting subjects settle first. diff --git a/pyproject.toml b/pyproject.toml index 9ab7233..f3a3e87 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "braindb" -version = "0.9.0" +version = "0.10.0" description = "Persistent memory for LLM agents — thoughts, facts, sources, and behavioral rules with fuzzy + semantic search, graph traversal, and an internal agent." readme = "README.md" license = "Apache-2.0" diff --git a/tests/test_handoff_hooks.py b/tests/test_handoff_hooks.py index d1b2345..5d6e00b 100644 --- a/tests/test_handoff_hooks.py +++ b/tests/test_handoff_hooks.py @@ -86,6 +86,61 @@ def __init__(self, s: str): assert _estimate_tokens(items) == 300 +def test_estimate_tokens_counts_tool_output_dict(): + """Tool results are `function_call_output` items: the payload sits + under `output`, never `content`. These dominate a writer's context + (section reads, recalls), so missing them left the estimate a + near-constant and the handoff nudge never fired.""" + items = [ + {"role": "user", "content": "x" * 400}, + {"type": "function_call_output", "call_id": "c1", "output": "y" * 800}, + ] + # 400 + 800 = 1200 chars / 4 = 300 tokens + assert _estimate_tokens(items) == 300 + + +def test_estimate_tokens_object_with_output_attr(): + """SDK item objects carrying `.output` (no `.content`) count too.""" + class FakeToolOutput: + output = "z" * 1200 + + assert _estimate_tokens([FakeToolOutput()]) == 300 + + +def test_estimate_tokens_counts_tool_call_arguments_dict(): + """An assistant TOOL CALL is a `function_call` item: its payload sits + under `arguments`, with neither `content` nor `output`. For a writer + this is the single largest term in the conversation — one + `edit_wiki_section` call carries a whole section — and it previously + counted as zero, so the estimate tracked only half the growth.""" + items = [ + {"role": "user", "content": "x" * 400}, + {"type": "function_call", "call_id": "c1", + "name": "edit_wiki_section", "arguments": "y" * 800}, + ] + # 400 + 800 = 1200 chars / 4 = 300 tokens + assert _estimate_tokens(items) == 300 + + +def test_estimate_tokens_object_with_arguments_attr(): + """SDK item objects carrying `.arguments` (no `.content`/`.output`).""" + class FakeToolCall: + arguments = "z" * 1200 + + assert _estimate_tokens([FakeToolCall()]) == 300 + + +def test_estimate_tokens_counts_call_and_result_together(): + """A realistic writer turn: the call it made plus the result it got + back. Both must count, or a section edit looks free.""" + items = [ + {"type": "function_call", "call_id": "c1", + "name": "edit_wiki_section", "arguments": "a" * 2000}, + {"type": "function_call_output", "call_id": "c1", "output": "b" * 2000}, + ] + assert _estimate_tokens(items) == 1000 + + def test_estimate_tokens_unknown_shape_contributes_zero(): """Unknown shapes (no recognisable text) must not raise. Lower-bound estimate is the safe side — we'd rather under-count than crash.""" diff --git a/tests/test_wiki_reconcile_dangling.py b/tests/test_wiki_reconcile_dangling.py new file mode 100644 index 0000000..9a3dd7f --- /dev/null +++ b/tests/test_wiki_reconcile_dangling.py @@ -0,0 +1,219 @@ +"""`reconcile_summarises_additive` must never let CONTENT block BOOKKEEPING. + +Exercises the function directly against the live Postgres instance, in the +style of `test_wiki_jobs_grouping.py`: seed a minimal wiki + entities, call +the function, assert, clean up in `try/finally`. + +The contract under test +----------------------- + +A wiki body cites entities inline as `[[ref:UUID]]`, and this function +mirrors each citation into a `wiki --summarises--> e` relation. It is the +step that makes a member stop being an orphan, and it runs in the SAME +transaction as `finish_jobs`. + +So a citation whose entity does not exist is not a content problem — it is +a liveness problem. Inserting it raises a foreign-key violation, which +aborts the whole transaction and takes `finish_jobs` with it. The section +edits made during the agent run are already committed, so the page keeps +growing while its jobs are never closed and re-queue forever. + +That is not hypothetical. A subagent without the section tools retyped a +52k-char body through `update_entity` to change one line and flipped one +hex digit of a cited UUID (`...-4e7c` -> `...-4f7c`). Every subsequent +write on that page returned 500, no job closed for three days, and the +`summarises` set froze at 114 while the body grew 44k -> 73.5k chars. + +The fix under test: a dangling ref is SKIPPED and REPORTED, never raised. +Reporting matters — the caller spreads the result into the `wiki_write` +activity log, so this stays visible rather than silent. +""" +from __future__ import annotations + +import os +import uuid +from typing import Iterator + +import psycopg2 +import pytest + +from braindb.services import wiki_jobs + + +# Same default as tests/conftest.py — the isolated stack from +# docker-compose.test.yml. An explicit DATABASE_URL in the env wins. +DB_URL = os.getenv( + "DATABASE_URL", "postgresql://braindb:braindb@localhost:5436/braindb_test" +) + + +# ---------------------------------------------------------------- helpers -- + + +def _insert_wiki(conn, label: str) -> str: + """Minimal wiki entity + its keyword + wikis_ext row (wikis_ext expects + member_keyword_ids non-empty). Returns the wiki entity UUID as text.""" + wid, kw_id = uuid.uuid4(), uuid.uuid4() + with conn.cursor() as cur: + cur.execute( + """INSERT INTO entities (id, entity_type, content, keywords, source, importance) + VALUES (%s, 'keyword', %s, %s, 'agent-inference', 0.5)""", + (str(kw_id), f"_pytest_reconcile_kw_{label}", [f"_pytest_reconcile_{label}"]), + ) + cur.execute( + """INSERT INTO entities (id, entity_type, content, keywords, source, importance) + VALUES (%s, 'wiki', %s, %s, 'agent-inference', 0.5)""", + (str(wid), f"# Test wiki ({label})\n", [f"_pytest_reconcile_{label}"]), + ) + cur.execute( + """INSERT INTO wikis_ext (entity_id, canonical_name, language, + member_keyword_ids, revision) + VALUES (%s, %s, 'en', %s::uuid[], 1)""", + (str(wid), f"PytestReconcile_{label}", [str(kw_id)]), + ) + return str(wid) + + +def _insert_fact(conn, label: str) -> str: + fid = uuid.uuid4() + with conn.cursor() as cur: + cur.execute( + """INSERT INTO entities (id, entity_type, content, keywords, source, importance) + VALUES (%s, 'fact', %s, %s, 'user-stated', 0.5)""", + (str(fid), f"_pytest_reconcile_fact_{label}", [f"_pytest_reconcile_{label}"]), + ) + return str(fid) + + +def _summarised_ids(conn, wiki_id: str) -> set[str]: + with conn.cursor() as cur: + cur.execute( + "SELECT to_entity_id::text FROM relations " + "WHERE from_entity_id = %s AND relation_type = 'summarises'", + (wiki_id,), + ) + return {r[0].lower() for r in cur.fetchall()} + + +def _cleanup(conn, entity_ids: list[str]) -> None: + with conn.cursor() as cur: + if entity_ids: + cur.execute("DELETE FROM entities WHERE id = ANY(%s::uuid[])", (entity_ids,)) + cur.execute( + "DELETE FROM entities WHERE entity_type='keyword' " + "AND content LIKE '_pytest_reconcile_kw_%'" + ) + + +def _body(*refs: str) -> str: + lines = "\n".join(f"- claim [[ref:{r}]]" for r in refs) + return f"\n{lines}\n" + + +@pytest.fixture +def db() -> Iterator[psycopg2.extensions.connection]: + """One autocommit psycopg2 connection per test, closed at teardown.""" + c = psycopg2.connect(DB_URL) + c.autocommit = True + try: + yield c + finally: + c.close() + + +# ------------------------------------------------------------------ tests -- + + +def test_live_refs_become_summarises_relations(db): + """Baseline: the normal path is unchanged.""" + wid = _insert_wiki(db, "live") + f1, f2 = _insert_fact(db, "live1"), _insert_fact(db, "live2") + try: + res = wiki_jobs.reconcile_summarises_additive(db, wid, _body(f1, f2)) + assert res["relations_added"] == 2 + assert res["refs_skipped"] == [] + assert _summarised_ids(db, wid) == {f1.lower(), f2.lower()} + finally: + _cleanup(db, [wid, f1, f2]) + + +def test_dangling_ref_does_not_raise(db): + """THE regression. A FK violation here aborts the caller's transaction + and `finish_jobs` never runs, so the jobs re-queue forever.""" + wid = _insert_wiki(db, "dangle") + ghost = str(uuid.uuid4()) # never inserted + try: + res = wiki_jobs.reconcile_summarises_additive(db, wid, _body(ghost)) + assert res["relations_added"] == 0 + assert res["refs_skipped"] == [ghost.lower()] + finally: + _cleanup(db, [wid]) + + +def test_one_dangling_ref_does_not_block_the_good_ones(db): + """The corrupted page had 3 bad refs among ~300 good ones. All the + valid citations must still be recorded.""" + wid = _insert_wiki(db, "mixed") + f1, f2 = _insert_fact(db, "mixed1"), _insert_fact(db, "mixed2") + ghost = str(uuid.uuid4()) + try: + res = wiki_jobs.reconcile_summarises_additive(db, wid, _body(f1, ghost, f2)) + assert res["relations_added"] == 2 + assert res["refs_skipped"] == [ghost.lower()] + assert _summarised_ids(db, wid) == {f1.lower(), f2.lower()} + finally: + _cleanup(db, [wid, f1, f2]) + + +def test_skipped_refs_are_reported_not_silent(db): + """The caller spreads this dict into the `wiki_write` activity log, so + a skipped ref stays visible — "never a silent bad write".""" + wid = _insert_wiki(db, "report") + g1, g2 = str(uuid.uuid4()), str(uuid.uuid4()) + try: + res = wiki_jobs.reconcile_summarises_additive(db, wid, _body(g1, g2)) + assert sorted(res["refs_skipped"]) == sorted([g1.lower(), g2.lower()]) + finally: + _cleanup(db, [wid]) + + +def test_reconcile_remains_additive_only(db): + """A pre-existing relation whose entity is no longer cited must NOT be + removed — the LLM owns retraction via `delete_relation`.""" + wid = _insert_wiki(db, "additive") + f1, f2 = _insert_fact(db, "add1"), _insert_fact(db, "add2") + try: + wiki_jobs.reconcile_summarises_additive(db, wid, _body(f1, f2)) + # now re-run with a body citing only f1 + res = wiki_jobs.reconcile_summarises_additive(db, wid, _body(f1)) + assert res["relations_added"] == 0 + assert res["relations_removed"] == 0 + assert _summarised_ids(db, wid) == {f1.lower(), f2.lower()} + finally: + _cleanup(db, [wid, f1, f2]) + + +def test_rerun_is_idempotent(db): + """Re-running on the same body adds nothing — the cron/retry path leans + on this.""" + wid = _insert_wiki(db, "idem") + f1 = _insert_fact(db, "idem1") + try: + first = wiki_jobs.reconcile_summarises_additive(db, wid, _body(f1)) + second = wiki_jobs.reconcile_summarises_additive(db, wid, _body(f1)) + assert first["relations_added"] == 1 + assert second["relations_added"] == 0 + assert _summarised_ids(db, wid) == {f1.lower()} + finally: + _cleanup(db, [wid, f1]) + + +def test_body_with_no_refs_is_a_clean_noop(db): + wid = _insert_wiki(db, "norefs") + try: + res = wiki_jobs.reconcile_summarises_additive( + db, wid, "\nprose with no citations\n") + assert res == {"relations_added": 0, "relations_removed": 0, + "refs_skipped": []} + finally: + _cleanup(db, [wid]) diff --git a/tests/test_wiki_sections.py b/tests/test_wiki_sections.py index 5b168e2..e63adea 100644 --- a/tests/test_wiki_sections.py +++ b/tests/test_wiki_sections.py @@ -2,9 +2,9 @@ splicing layer behind the writer's section-edit tools. These tests cover the DB-free functions only (`parse_sections`, -`splice_section`, `delete_section`, `check_grammar`). The DB helpers -(`fetch_wiki_for_section_op`, `apply_section_write`) are covered by -the end-to-end smoke test inside `braindb_api` (see plan Phase 1). +`splice_section`, `append_to_section`, `delete_section`, `check_grammar`). +The DB helpers (`fetch_wiki_for_section_op`, `apply_section_write`) are +covered by the end-to-end smoke test inside `braindb_api` (see plan Phase 1). The contract being tested: @@ -14,6 +14,11 @@ - `splice_section` REPLACES an existing section's content, or APPENDS a fresh section if the name is new. Bytes outside the targeted section are preserved exactly. +- `append_to_section` ADDS to an existing section without the caller + supplying its prior content — the genuinely-additive "one more + citation" case. Existing content is preserved at the content level + (trailing blank lines collapse to one; markers re-emit canonically): + no claim or citation can be lost. - `delete_section` removes a section, raises `KeyError` if missing. - `check_grammar` flags: no markers, malformed `[[ref:` tokens, missing Summary callout. Tolerates the grouped-refs variant `[[ref:UUID1], @@ -28,9 +33,11 @@ from braindb.services.wiki_sections import ( Section, StaleRevisionError, + append_to_section, check_grammar, delete_section, parse_sections, + replace_header, splice_section, ) @@ -165,6 +172,61 @@ def test_delete_section_preserves_header(): assert new_header == original_header +# ====================================================================== # +# replace_header — the header becomes editable # +# ====================================================================== # +# +# Section tools could edit every section but never the block above the +# first marker, so a page's Summary could permanently contradict its own +# body (observed live: "is applying" against a recorded acceptance). +# `replace_header` closes that; these tests pin that it touches ONLY the +# header. + +NEW_HEADER = ( + "\n" + "# Test\n" + "> **Summary:** updated one line\n" + "> **Disambiguation:** what this is, updated\n" +) + + +def test_replace_header_replaces_only_the_header(): + out = replace_header(NORMAL_BODY, NEW_HEADER) + header, sections = parse_sections(out) + assert "> **Summary:** updated one line" in header + assert "revision=1" not in header # the stale token is gone with the old header + before = {s.name: s.content for s in parse_sections(NORMAL_BODY)[1]} + after = {s.name: s.content for s in sections} + assert after == before + + +def test_replace_header_keeps_section_order(): + out = replace_header(NORMAL_BODY, NEW_HEADER) + assert [s.name for s in parse_sections(out)[1]] == [ + "overview", "timeline", "references"] + + +def test_replace_header_result_is_grammar_clean(): + assert check_grammar(replace_header(NORMAL_BODY, NEW_HEADER)) == [] + + +def test_replace_header_without_trailing_newline(): + out = replace_header(NORMAL_BODY, NEW_HEADER.rstrip("\n")) + header, sections = parse_sections(out) + assert header.endswith("\n") # _rebuild normalises + assert len(sections) == 3 + + +def test_replace_header_empty_header_drops_it(): + """Pin current `_rebuild` behaviour: an empty header is omitted, the + body then starts at the first marker.""" + out = replace_header(NORMAL_BODY, "") + header, sections = parse_sections(out) + assert header == "" + assert out.startswith("") + assert len(sections) == 3 + + # ====================================================================== # # Round-trip identity # # ====================================================================== # @@ -196,6 +258,19 @@ def test_grammar_flags_missing_markers(): assert any("no \nprose\n" + ) + issues = check_grammar(body) + assert any("wiki:meta" in i for i in issues) + # and only that — Summary and markers are fine here + assert not any("Summary" in i for i in issues) + + def test_grammar_flags_missing_summary(): body = ( "\n" @@ -262,3 +337,79 @@ def test_section_is_frozen_dataclass(): def test_section_char_count_property(): s = Section(name="x", content="abcdef") assert s.char_count == 6 + + +# ====================================================================== # +# append_to_section — the incremental add # +# ====================================================================== # +# +# The pipeline's real unit of work is "add one [[ref:UUID]] bullet". Doing +# that through `splice_section` requires the caller to re-emit the section's +# WHOLE content, which on a section past the read cap cannot be done +# correctly at all. These tests pin the property that makes appending safe: +# whatever was already in the section survives byte-for-byte, without the +# caller ever having to hold it. + +def test_append_preserves_existing_content_byte_for_byte(): + before = next(s for s in parse_sections(NORMAL_BODY)[1] + if s.name == "references") + out = append_to_section(NORMAL_BODY, "references", "- [[ref:%s]] — source C" % UUID_A) + after = next(s for s in parse_sections(out)[1] if s.name == "references") + # every original line is still present, in order, unmodified + original_lines = [ln for ln in before.content.splitlines() if ln.strip()] + after_lines = [ln for ln in after.content.splitlines() if ln.strip()] + assert after_lines[:len(original_lines)] == original_lines + + +def test_append_adds_the_new_text_at_the_end(): + out = append_to_section(NORMAL_BODY, "references", "- new tail line") + section = next(s for s in parse_sections(out)[1] if s.name == "references") + assert section.content.rstrip("\n").endswith("- new tail line") + + +def test_append_does_not_touch_other_sections_or_header(): + out = append_to_section(NORMAL_BODY, "references", "- new tail line") + hdr_before, secs_before = parse_sections(NORMAL_BODY) + hdr_after, secs_after = parse_sections(out) + assert hdr_after == hdr_before + untouched_before = {s.name: s.content for s in secs_before if s.name != "references"} + untouched_after = {s.name: s.content for s in secs_after if s.name != "references"} + assert untouched_after == untouched_before + + +def test_append_keeps_section_order(): + out = append_to_section(NORMAL_BODY, "overview", "more prose") + assert [s.name for s in parse_sections(out)[1]] == [ + "overview", "timeline", "references"] + + +def test_append_to_missing_section_creates_it_like_splice(): + out = append_to_section(NORMAL_BODY, "sources", "narrative provenance") + names = [s.name for s in parse_sections(out)[1]] + assert names == ["overview", "timeline", "references", "sources"] + created = next(s for s in parse_sections(out)[1] if s.name == "sources") + assert "narrative provenance" in created.content + + +def test_append_result_is_reparseable_normal_form(): + out = append_to_section(NORMAL_BODY, "references", "- another") + # a second append must behave identically on the result of the first + out2 = append_to_section(out, "references", "- and another") + section = next(s for s in parse_sections(out2)[1] if s.name == "references") + assert "- another" in section.content + assert "- and another" in section.content + assert check_grammar(out2) == [] + + +def test_append_never_loses_refs_on_a_large_section(): + """The regression this exists for: a section far bigger than the tool + read cap (8000) must survive an append intact, because the caller never + supplies its prior content.""" + big = "\n".join(f"- [[ref:{UUID_A}]] — line {i}" for i in range(1200)) + body = splice_section(NORMAL_BODY, "references", big) + assert len(next(s for s in parse_sections(body)[1] + if s.name == "references").content) > 8000 + out = append_to_section(body, "references", f"- [[ref:{UUID_B}]] — new") + after = next(s for s in parse_sections(out)[1] if s.name == "references") + assert after.content.count("[[ref:") == 1201 + assert "line 0" in after.content and "line 1199" in after.content diff --git a/tests/test_wiki_selfheal_db.py b/tests/test_wiki_selfheal_db.py new file mode 100644 index 0000000..879d548 --- /dev/null +++ b/tests/test_wiki_selfheal_db.py @@ -0,0 +1,580 @@ +"""Self-healing and guard behaviour, tested against the real database. + +Every test here pins a property that would NOT fail under a revert if it +were asserted on strings or private tuples (the earlier, weaker tests did +exactly that). These call the real functions and the real tools. + +The contract under test +----------------------- + +1. THE SELF-HEAL LOOP. A job past both the lease and the reclaim ceiling is + un-claimable, and `assigned` counts as an ACTIVE job in + `_orphan_conditions()` — so without a disposition its entity would be + permanently barred from re-triage: silent, unbounded loss. `run_cron` + must flip such rows to `failed` (the status that, by design, returns the + entity to the orphan pool) and re-enqueue a fresh triage job for the + entity — in the same tick. + +2. THE RECLAIM CEILING, behaviourally: `claim_jobs` re-claims a + lease-expired job below the ceiling and refuses one at it. + +3. `update_entity` GUARDS, through the real tool (`on_invoke_tool` with a + stub ctx): a wiki body cannot be overwritten, a fact still can, other + fields still apply to wikis, and `content=""` never silently destroys a + body. NOTE: the SDK converts tool exceptions into an error STRING, so + every assertion here is on reply content, never on "no exception". + +4. ONE CITATION PREDICATE. `uncited_members` splits un-cited members into + `missing` (entity exists — real work) and `gone` (deleted since triage — + can never be cited, must not wedge the job). The router's no-op gate and + the `check_members_cited` tool both call it, so agreement is by + construction; these tests pin the split itself and the tool's reporting. +""" +from __future__ import annotations + +import asyncio +import json +import os +import uuid +from typing import Iterator + +import psycopg2 +import psycopg2.extras +import pytest + +from braindb.services import wiki_jobs + + +# Same default as tests/conftest.py — the isolated stack from +# docker-compose.test.yml. An explicit DATABASE_URL in the environment wins +# (in CI it points at the workflow's Postgres service container). +DB_URL = os.getenv( + "DATABASE_URL", "postgresql://braindb:braindb@localhost:5436/braindb_test" +) + + +class _Ctx: + """Minimal ToolContext stub: the SDK's invocation path reads only + `tool_name` (agents/tool.py `_on_invoke_tool_impl`).""" + tool_name = "test" + + +def _invoke(tool, **kwargs) -> str: + """Run a FunctionTool the way the SDK would, returning its string reply.""" + return asyncio.run(tool.on_invoke_tool(_Ctx(), json.dumps(kwargs))) + + +# ---------------------------------------------------------------- helpers -- + + +def _insert_entity(conn, entity_type: str, label: str, *, + age_minutes: int = 0, notes: str | None = None, + importance: float = 0.5) -> str: + eid = uuid.uuid4() + with conn.cursor() as cur: + cur.execute( + """INSERT INTO entities (id, entity_type, content, keywords, source, + importance, notes, created_at) + VALUES (%s, %s, %s, %s, 'user-stated', %s, %s, + now() - make_interval(mins => %s))""", + (str(eid), entity_type, f"_pytest_selfheal_{label}", + [f"_pytest_selfheal_{label}"], importance, notes, age_minutes), + ) + return str(eid) + + +def _insert_wiki(conn, label: str, body: str) -> str: + wid, kw_id = uuid.uuid4(), uuid.uuid4() + with conn.cursor() as cur: + cur.execute( + """INSERT INTO entities (id, entity_type, content, keywords, source, importance) + VALUES (%s, 'keyword', %s, %s, 'agent-inference', 0.5)""", + (str(kw_id), f"_pytest_selfheal_kw_{label}", [f"_pytest_selfheal_{label}"]), + ) + cur.execute( + """INSERT INTO entities (id, entity_type, content, keywords, source, importance) + VALUES (%s, 'wiki', %s, %s, 'agent-inference', 0.5)""", + (str(wid), body, [f"_pytest_selfheal_{label}"]), + ) + cur.execute( + """INSERT INTO wikis_ext (entity_id, canonical_name, language, + member_keyword_ids, revision) + VALUES (%s, %s, 'en', %s::uuid[], 1)""", + (str(wid), f"PytestSelfheal_{label}", [str(kw_id)]), + ) + return str(wid) + + +def _insert_job(conn, *, job_type: str, entity_ids: list[str], + status: str = "pending", attempts: int = 0, + assigned_age_minutes: int | None = None, + created_age_minutes: int = 0, + target_wiki_id: str | None = None, + dedupe_key: str | None = None) -> str: + jid = uuid.uuid4() + dedupe = dedupe_key or f"_pytest_selfheal_{job_type}_{uuid.uuid4().hex}" + with conn.cursor() as cur: + cur.execute( + """INSERT INTO wiki_job + (id, job_type, status, target_wiki_id, entity_ids, dedupe_key, + attempts, assigned_at, created_at, rationale) + VALUES (%s, %s, %s, %s, %s::uuid[], %s, %s, + CASE WHEN %s::int IS NULL THEN NULL + ELSE now() - make_interval(mins => %s) END, + now() - make_interval(mins => %s), + 'pytest selfheal')""", + (str(jid), job_type, status, target_wiki_id, entity_ids, dedupe, + attempts, assigned_age_minutes, assigned_age_minutes or 0, + created_age_minutes), + ) + return str(jid) + + +def _job_row(conn, job_id: str) -> dict: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT * FROM wiki_job WHERE id = %s", (job_id,)) + return dict(cur.fetchone()) + + +def _cleanup(conn, *, entity_ids: list[str] = (), job_ids: list[str] = ()) -> None: + with conn.cursor() as cur: + if job_ids: + cur.execute("DELETE FROM wiki_job WHERE id = ANY(%s::uuid[])", (list(job_ids),)) + cur.execute("DELETE FROM wiki_job WHERE rationale = 'pytest selfheal'") + cur.execute("DELETE FROM wiki_job WHERE dedupe_key LIKE 'triage:%%' " + "AND entity_ids && %s::uuid[]", + (list(entity_ids) or ["00000000-0000-0000-0000-000000000000"],)) + if entity_ids: + cur.execute("DELETE FROM entities WHERE id = ANY(%s::uuid[])", (list(entity_ids),)) + cur.execute("DELETE FROM entities WHERE content LIKE '_pytest_selfheal_%%' " + "OR (entity_type='keyword' AND content LIKE '_pytest_selfheal_kw_%%')") + + +@pytest.fixture +def db() -> Iterator[psycopg2.extensions.connection]: + c = psycopg2.connect(DB_URL) + c.autocommit = True + try: + yield c + finally: + c.close() + + +@pytest.fixture(autouse=True) +def _point_tools_at_test_db(monkeypatch): + """The tools open their own connections via `get_conn()`, which reads + `settings.database_url` per call — point it at the test database.""" + from braindb.config import settings + monkeypatch.setattr(settings, "database_url", DB_URL) + + +# ------------------------------------------------- 1. the self-heal loop -- + + +def test_wedged_job_is_failed_and_entity_retriaged_in_one_cron_tick(db): + """THE self-heal test. Without run_cron's disposition, this job would + be un-claimable forever and its entity barred from re-triage for good.""" + eid = _insert_entity(db, "fact", "wedge", age_minutes=90) # settled + jid = _insert_job(db, job_type="triage", entity_ids=[eid], + status="assigned", + attempts=wiki_jobs.ASSIGNED_MAX_RECLAIMS, + assigned_age_minutes=wiki_jobs.ASSIGNED_LEASE_MIN + 60, + dedupe_key=f"triage:{eid}") + try: + result = wiki_jobs.run_cron(db) + row = _job_row(db, jid) + assert row["status"] == "failed" + assert "reclaim ceiling" in (row["last_error"] or "") + assert result["assigned_expired_failed"] >= 1 + # Same tick: the entity is an orphan again and a FRESH pending + # triage job exists (the partial dedupe index ignores failed rows). + with db.cursor() as cur: + cur.execute( + "SELECT count(*) FROM wiki_job WHERE dedupe_key = %s " + "AND status = 'pending'", (f"triage:{eid}",)) + assert cur.fetchone()[0] == 1 + finally: + _cleanup(db, entity_ids=[eid], job_ids=[jid]) + + +def test_healthy_assigned_job_is_left_alone_by_cron(db): + """A job under the ceiling, or within its lease, is someone's live work.""" + eid = _insert_entity(db, "fact", "healthy", age_minutes=90) + under_ceiling = _insert_job(db, job_type="triage", entity_ids=[eid], + status="assigned", attempts=1, + assigned_age_minutes=wiki_jobs.ASSIGNED_LEASE_MIN + 60) + in_lease = _insert_job(db, job_type="triage", entity_ids=[eid], + status="assigned", + attempts=wiki_jobs.ASSIGNED_MAX_RECLAIMS, + assigned_age_minutes=1) + try: + wiki_jobs.run_cron(db) + assert _job_row(db, under_ceiling)["status"] == "assigned" + assert _job_row(db, in_lease)["status"] == "assigned" + finally: + _cleanup(db, entity_ids=[eid], job_ids=[under_ceiling, in_lease]) + + +# ------------------------------------------ 2. the ceiling, behaviourally -- + + +def test_claim_jobs_reclaims_below_ceiling_refuses_at_it(db): + jid = _insert_job(db, job_type="attach", entity_ids=[], + status="assigned", + attempts=wiki_jobs.ASSIGNED_MAX_RECLAIMS - 1, + assigned_age_minutes=wiki_jobs.ASSIGNED_LEASE_MIN + 60) + try: + # Below the ceiling + lease expired -> reclaimable. + assert wiki_jobs.claim_jobs(db, [jid]) == 1 + assert _job_row(db, jid)["attempts"] == wiki_jobs.ASSIGNED_MAX_RECLAIMS + # Now AT the ceiling: expire the lease again and try to re-claim. + with db.cursor() as cur: + cur.execute( + "UPDATE wiki_job SET assigned_at = now() - make_interval(" + "mins => %s) WHERE id = %s", + (wiki_jobs.ASSIGNED_LEASE_MIN + 60, jid)) + assert wiki_jobs.claim_jobs(db, [jid]) == 0 + finally: + _cleanup(db, job_ids=[jid]) + + +# --------------------------------------- 3. update_entity, the real tool -- + + +def test_update_entity_tool_refuses_wiki_body_and_leaves_db_untouched(db): + from braindb.agent.tools import update_entity + body = "\noriginal wiki body\n" + wid = _insert_wiki(db, "guard", body) + try: + reply = _invoke(update_entity, entity_id=wid, content="OVERWRITTEN") + assert "content ignored" in reply + assert "edit_wiki_section" in reply + with db.cursor() as cur: + cur.execute("SELECT content FROM entities WHERE id = %s", (wid,)) + assert cur.fetchone()[0] == body + finally: + _cleanup(db, entity_ids=[wid]) + + +def test_update_entity_tool_still_applies_other_fields_to_a_wiki(db): + from braindb.agent.tools import update_entity + wid = _insert_wiki(db, "fields", "\nbody\n") + try: + reply = _invoke(update_entity, entity_id=wid, notes="curator note", + importance=0.9) + assert reply.startswith("Updated entity") + with db.cursor() as cur: + cur.execute("SELECT notes, importance FROM entities WHERE id = %s", (wid,)) + notes, importance = cur.fetchone() + assert notes == "curator note" + assert float(importance) == 0.9 + finally: + _cleanup(db, entity_ids=[wid]) + + +def test_update_entity_tool_still_updates_a_fact_body(db): + from braindb.agent.tools import update_entity + eid = _insert_entity(db, "fact", "editable") + try: + reply = _invoke(update_entity, entity_id=eid, content="corrected fact") + assert reply.startswith("Updated entity") + assert "ignored" not in reply + with db.cursor() as cur: + cur.execute("SELECT content FROM entities WHERE id = %s", (eid,)) + assert cur.fetchone()[0] == "corrected fact" + finally: + _cleanup(db, entity_ids=[eid]) + + +def test_update_entity_tool_never_silently_blanks_a_body(db): + """Observed live: content='' wiped a thought whose ref was already baked + into a wiki. Blanking must be a warned no-op, not silent destruction.""" + from braindb.agent.tools import update_entity + eid = _insert_entity(db, "thought", "precious", notes="keep me") + try: + reply = _invoke(update_entity, entity_id=eid, content="") + assert "empty content ignored" in reply + with db.cursor() as cur: + cur.execute("SELECT content FROM entities WHERE id = %s", (eid,)) + assert cur.fetchone()[0] == "_pytest_selfheal_precious" + finally: + _cleanup(db, entity_ids=[eid]) + + +# ------------------------------------------- 4. one citation predicate -- + + +def test_uncited_members_splits_missing_from_gone(db): + f_cited = _insert_entity(db, "fact", "cited") + f_uncited = _insert_entity(db, "fact", "uncited") + ghost = str(uuid.uuid4()) # never inserted — deleted-since-triage case + body = f"claim [[ref:{f_cited}]]\n" + try: + missing, gone = wiki_jobs.uncited_members( + db, body, [f_cited, f_uncited, ghost]) + assert missing == [f_uncited] # real outstanding work + assert gone == [ghost] # can never be cited — must not wedge + finally: + _cleanup(db, entity_ids=[f_cited, f_uncited]) + + +def test_uncited_members_all_cited_is_clean(db): + f1 = _insert_entity(db, "fact", "done1") + try: + assert wiki_jobs.uncited_members(db, f"x [[ref:{f1}]]", [f1]) == ([], []) + finally: + _cleanup(db, entity_ids=[f1]) + + +def test_check_members_cited_tool_reports_gone_distinctly(db): + """The tool and the router share `uncited_members`, so agreement is by + construction; what the tool adds is honest reporting — a deleted member + must show as `gone`, not as outstanding work.""" + from braindb.agent.tools import check_members_cited + f_cited = _insert_entity(db, "fact", "tool_cited") + f_uncited = _insert_entity(db, "fact", "tool_uncited") + ghost = str(uuid.uuid4()) + wid = _insert_wiki(db, "tool", f"\nx [[ref:{f_cited}]]\n") + try: + reply = _invoke(check_members_cited, wiki_id=wid, + entity_ids=[f_cited, f_uncited, ghost]) + assert "cited: 1/3" in reply + not_cited_line = next(l for l in reply.splitlines() + if l.startswith("NOT_cited:")) + gone_line = next(l for l in reply.splitlines() if l.startswith("gone")) + assert f_uncited in not_cited_line and ghost not in not_cited_line + assert ghost in gone_line + finally: + _cleanup(db, entity_ids=[f_cited, f_uncited, wid]) + + +# --------------------------------------------- 5. the header capability -- + + +HDR_BODY = ( + "\n" + "# HdrTest\n" + "> **Summary:** old summary line\n" + "> **Disambiguation:** old scope\n" + "\nprose stays untouched\n" +) +NEW_HDR = ( + "\n" + "# HdrTest\n" + "> **Summary:** new summary line\n" + "> **Disambiguation:** new scope\n" +) + + +def test_header_replace_edits_only_the_header_and_bumps_revision(db): + from braindb.agent.tools import edit_wiki_section + from braindb.services.wiki_jobs import extract_summary_disambig + wid = _insert_wiki(db, "hdr", HDR_BODY) + try: + reply = _invoke(edit_wiki_section, wiki_id=wid, section_name="header", + new_content=NEW_HDR, expect_revision=1) + assert reply.startswith("ok"), reply + assert "replaced" in reply + with db.cursor() as cur: + cur.execute("SELECT e.content, w.revision FROM entities e " + "JOIN wikis_ext w ON w.entity_id = e.id WHERE e.id = %s", + (wid,)) + content, revision = cur.fetchone() + assert "new summary line" in content + assert "old summary line" not in content + assert "prose stays untouched" in content # sections untouched + assert revision == 2 + # The existing persist path re-extracts from the body — prove the + # input it would read now carries the new header values. + summary, disambig = extract_summary_disambig(content) + assert summary == "new summary line" + assert disambig == "new scope" + finally: + _cleanup(db, entity_ids=[wid]) + + +def test_header_append_is_refused(db): + from braindb.agent.tools import edit_wiki_section + wid = _insert_wiki(db, "hdrapp", HDR_BODY) + try: + reply = _invoke(edit_wiki_section, wiki_id=wid, section_name="header", + new_content="extra", expect_revision=1, mode="append") + assert 'cannot append to "header"' in reply + finally: + _cleanup(db, entity_ids=[wid]) + + +def test_header_delete_is_refused(db): + from braindb.agent.tools import delete_wiki_section + wid = _insert_wiki(db, "hdrdel", HDR_BODY) + try: + reply = _invoke(delete_wiki_section, wiki_id=wid, + section_name="header", expect_revision=1) + assert "cannot be deleted" in reply + finally: + _cleanup(db, entity_ids=[wid]) + + +def test_header_replace_cas_rejects_stale_revision(db): + from braindb.agent.tools import edit_wiki_section + wid = _insert_wiki(db, "hdrcas", HDR_BODY) + try: + reply = _invoke(edit_wiki_section, wiki_id=wid, section_name="header", + new_content=NEW_HDR, expect_revision=99) + assert "stale revision" in reply + finally: + _cleanup(db, entity_ids=[wid]) + + +def test_outline_and_read_surface_the_header(db): + from braindb.agent.tools import read_wiki_outline, read_wiki_section + wid = _insert_wiki(db, "hdrout", HDR_BODY) + try: + outline = _invoke(read_wiki_outline, wiki_id=wid) + assert "header:" in outline and 'section "header"' in outline + read = _invoke(read_wiki_section, wiki_id=wid, section_name="header") + assert "section: header" in read + assert "old summary line" in read + assert "prose stays untouched" not in read # header only + finally: + _cleanup(db, entity_ids=[wid]) + + +def test_wiki_not_found_message_is_uniform_and_actionable(db): + """Models will always mistype UUIDs; the general recovery is the same + for every tool: copy the id exactly. One message, six tools.""" + from braindb.agent import tools as t + ghost = str(uuid.uuid4()) + replies = [ + _invoke(t.read_wiki_outline, wiki_id=ghost), + _invoke(t.read_wiki_section, wiki_id=ghost, section_name="overview"), + _invoke(t.check_members_cited, wiki_id=ghost, entity_ids=[ghost]), + _invoke(t.edit_wiki_section, wiki_id=ghost, section_name="overview", + new_content="x", expect_revision=1), + _invoke(t.delete_wiki_section, wiki_id=ghost, section_name="overview", + expect_revision=1), + _invoke(t.validate_wiki, wiki_id=ghost), + ] + for r in replies: + assert "wiki not found" in r, r + assert "Copy the wiki id" in r, r + + +# ------------------------------------- 6. restart-orphan release (A2) -- + + +def test_release_stale_assigned_releases_only_pre_cutoff(db): + from datetime import datetime, timedelta, timezone + old = _insert_job(db, job_type="attach", entity_ids=[], + status="assigned", attempts=2, assigned_age_minutes=30) + fresh = _insert_job(db, job_type="attach", entity_ids=[], + status="assigned", attempts=1, assigned_age_minutes=1) + try: + cutoff = datetime.now(timezone.utc) - timedelta(minutes=5) + released = wiki_jobs.release_stale_assigned(db, cutoff) + assert released >= 1 + old_row, fresh_row = _job_row(db, old), _job_row(db, fresh) + assert old_row["status"] == "pending" + assert old_row["assigned_at"] is None + assert old_row["attempts"] == 2 # preserved; claim increments later + assert fresh_row["status"] == "assigned" # after cutoff: live work + finally: + _cleanup(db, job_ids=[old, fresh]) + + +# --------------------------------------- 7. create dedupe by name (C) -- + + +def test_create_dedupe_collapses_same_name_while_active(db): + e1 = _insert_entity(db, "fact", "dupname1") + e2 = _insert_entity(db, "fact", "dupname2") + k1 = wiki_jobs.suggestion_dedupe_key( + "create", None, [e1], [], proposed_name="Foo Bar") + k2 = wiki_jobs.suggestion_dedupe_key( + "create", None, [e2], [], proposed_name="foo bar") + assert k1 == k2 # name-keyed, case-folded; seed ids irrelevant + try: + first = wiki_jobs.insert_suggestion( + db, job_type="create", target_wiki_id=None, entity_ids=[e1], + dedupe_key=k1, rationale="pytest selfheal", + proposed_name="Foo Bar", batch_id=None) + assert first is not None + second = wiki_jobs.insert_suggestion( + db, job_type="create", target_wiki_id=None, entity_ids=[e2], + dedupe_key=k2, rationale="pytest selfheal", + proposed_name="foo bar", batch_id=None) + assert second is None # collapsed while the first is active + # After the first completes, the partial index no longer spans it — + # a later create for the same name may insert again. + with db.cursor() as cur: + cur.execute("UPDATE wiki_job SET status='done' WHERE id = %s", + (first,)) + third = wiki_jobs.insert_suggestion( + db, job_type="create", target_wiki_id=None, entity_ids=[e2], + dedupe_key=k2, rationale="pytest selfheal", + proposed_name="foo bar", batch_id=None) + assert third is not None + finally: + _cleanup(db, entity_ids=[e1, e2]) + + +# ----------------------- 8. router-level: snapshot at claim + fail log -- + + +def test_writer_claim_snapshots_before_any_persist_and_failure_is_logged( + db, monkeypatch): + """THE reversibility test. The agent run is made to die immediately; + the pre-run snapshot must already exist (claim-time write), and the + crash must leave a visible wiki_write failure row (A3).""" + import braindb.routers.wiki as wiki_router + + wid = _insert_wiki(db, "snap", HDR_BODY) + jid = _insert_job(db, job_type="attach", entity_ids=[], + target_wiki_id=wid, created_age_minutes=10) + try: + async def boom(*a, **k): + raise RuntimeError("pytest-forced writer crash") + monkeypatch.setattr(wiki_router, "run_typed", boom) + res = asyncio.run(wiki_router.wiki_write()) + assert res["written"] == 0 + assert "pytest-forced" in res["reason"] + with db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + "SELECT operation, details FROM activity_log " + "WHERE entity_id = %s ORDER BY timestamp", (wid,)) + rows = cur.fetchall() + ops = [r["operation"] for r in rows] + assert "wiki_revise" in ops # snapshot at CLAIM; run died after + snap = next(r for r in rows if r["operation"] == "wiki_revise") + assert snap["details"]["from_revision"] == 1 + assert "prose stays untouched" in snap["details"]["prior_content"] + fail = next(r for r in rows if r["operation"] == "wiki_write") + assert fail["details"]["result"] in ("pending", "requeued", "failed") + assert "pytest-forced" in fail["details"]["error"] + finally: + _cleanup(db, entity_ids=[wid], job_ids=[jid]) + + +def test_maintainer_crash_is_logged_and_job_failed(db, monkeypatch): + import braindb.routers.wiki as wiki_router + + eid = _insert_entity(db, "fact", "mfail", age_minutes=90, importance=0.99) + jid = _insert_job(db, job_type="triage", entity_ids=[eid], + dedupe_key=f"triage:{eid}") + try: + async def boom(*a, **k): + raise RuntimeError("pytest-forced maintainer crash") + monkeypatch.setattr(wiki_router, "run_typed", boom) + res = asyncio.run(wiki_router.wiki_maintain()) + assert res["result"] == "failed" + assert _job_row(db, jid)["status"] == "failed" + with db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + "SELECT details FROM activity_log WHERE operation = " + "'wiki_maintain' AND details->>'result' = 'failed' " + "AND details->'jobs' @> to_jsonb(%s::text)", (jid,)) + row = cur.fetchone() + assert row is not None, "crashed maintainer left no activity row" + assert "pytest-forced" in row["details"]["error"] + finally: + _cleanup(db, entity_ids=[eid], job_ids=[jid]) diff --git a/tests/test_wiki_writer_guards.py b/tests/test_wiki_writer_guards.py new file mode 100644 index 0000000..675f408 --- /dev/null +++ b/tests/test_wiki_writer_guards.py @@ -0,0 +1,305 @@ +"""Unit tests for the guards that keep the wiki writer loop terminating. + +DB-free. Each test pins one invariant that a live 70-hour run proved was +NOT holding, so a regression here reproduces a failure we have actually +paid for: + +- `_claimable()` bounds the LEASE-RECLAIM branch (behavioural proof lives + in `test_wiki_selfheal_db.py`; the string tests here are cheap + documentation of the predicate's shape). +- The subagent/writer TOOLSETS are asserted on the BUILT agents, so + dropping `extra_tools=` from a factory fails these tests — asserting the + private tuples alone would not. +- Delegation depth is per-run-context: two CONCURRENT delegations must both + proceed (a process-global counter refuses the second — the exact live + failure: two sibling delegations 1.8ms apart, the second rejected), while + a NESTED delegation is still refused. +- `update_entity`'s protected-content map covers wikis (behavioural proof + through the real tool is in `test_wiki_selfheal_db.py`). + +These are guards, not behaviour: none of them judges or shapes content. +The citation predicate itself (`uncited_members`) is DB-backed and tested +in `test_wiki_selfheal_db.py` — the router and the `check_members_cited` +tool both call that single helper, so they cannot drift. +""" +from __future__ import annotations + +import asyncio +import json + +import braindb.agent.agent as agent_mod +import braindb.agent.tools as tools_mod +from braindb.services.wiki_jobs import ( + ASSIGNED_LEASE_MIN, + ASSIGNED_MAX_RECLAIMS, + _claimable, +) + + +def _agent_tool_names(agent) -> set[str]: + return {getattr(t, "name", None) for t in agent.tools} + + +# ====================================================================== # +# _claimable — the reclaim ceiling (shape; behaviour is DB-tested) # +# ====================================================================== # + +def test_claimable_bounds_the_reclaim_branch_by_attempts(): + """Without this the abandoned-run path is unbounded.""" + sql = _claimable() + assert f"attempts < {ASSIGNED_MAX_RECLAIMS}" in sql + + +def test_claimable_still_admits_pending_unconditionally(): + """A fresh job must never be gated by the reclaim ceiling — the cap + applies only to the branch nothing else bounds.""" + sql = _claimable() + pending_clause, _, reclaim_clause = sql.partition(" OR ") + assert "status = 'pending'" in pending_clause + assert "attempts" not in pending_clause + assert "attempts" in reclaim_clause + + +def test_claimable_still_requires_lease_expiry_to_reclaim(): + sql = _claimable() + assert "assigned_at <" in sql + assert f"mins => {ASSIGNED_LEASE_MIN}" in sql + + +def test_claimable_alias_qualifies_every_column(): + """The predicate is interpolated into queries that alias the table; + an unqualified column there is a runtime SQL error.""" + sql = _claimable("j") + for col in ("status", "attempts", "assigned_at"): + assert f"j.{col}" in sql + + +def test_reclaim_ceiling_sits_above_the_graceful_cap(): + """`release_or_fail_jobs` uses max_attempts=3. The reclaim ceiling must + be higher, or it would pre-empt normal failure handling.""" + assert ASSIGNED_MAX_RECLAIMS > 3 + + +# ====================================================================== # +# update_entity — protected-content map # +# ====================================================================== # + +def test_content_readonly_covers_wiki_and_datasource(): + assert set(tools_mod._CONTENT_READONLY) == {"datasource", "wiki"} + + +def test_wiki_readonly_message_points_at_the_section_tools(): + """The model has to be told where to go, or it reaches for search_sql.""" + msg = tools_mod._CONTENT_READONLY["wiki"] + assert "edit_wiki_section" in msg + + +def test_datasource_message_is_unchanged(): + """Pre-existing behaviour must not drift while generalising the guard.""" + assert tools_mod._CONTENT_READONLY["datasource"] == ( + "datasource bodies are read-only; use notes for analysis" + ) + + +# ====================================================================== # +# Toolsets — asserted on the BUILT agents, not the private tuples # +# ====================================================================== # + +def test_built_subagent_carries_the_wiki_read_tools(): + """Without these a subagent cannot inspect a page the safe way, and + falls back to paging the raw body and re-emitting it through + update_entity — which is how a cited UUID got corrupted live.""" + names = _agent_tool_names(agent_mod.get_subagent()) + assert {"read_wiki_outline", "read_wiki_section", + "check_members_cited", "validate_wiki"} <= names + + +def test_built_subagent_has_no_wiki_write_tools(): + """One writer per wiki is what makes the revision CAS meaningful, and a + subagent cannot hand off, so it has no business holding a revision.""" + names = _agent_tool_names(agent_mod.get_subagent()) + assert "edit_wiki_section" not in names + assert "delete_wiki_section" not in names + assert "handoff_to_successor" not in names + + +def test_built_writer_keeps_every_tool_it_had(): + """Regression guard: the writer's toolset may grow, never shrink.""" + names = _agent_tool_names(agent_mod.get_writer_agent()) + assert {"read_wiki_outline", "read_wiki_section", "check_members_cited", + "edit_wiki_section", "delete_wiki_section", "validate_wiki", + "handoff_to_successor"} <= names + + +def test_delegate_docstring_no_longer_promises_the_full_toolset(): + """The false promise is what made writers delegate edits a subagent + could not perform.""" + doc = tools_mod.delegate_to_subagent.description or "" + assert "all the same BrainDB tools" not in doc + assert "read_wiki_section" in doc + + +# ====================================================================== # +# Delegation depth — per run-context, behaviourally # +# ====================================================================== # + +class _Ctx: + """Minimal ToolContext stub — the SDK invocation path reads only + `tool_name`.""" + tool_name = "delegate_to_subagent" + + +def test_concurrent_delegations_proceed_and_nested_is_refused(monkeypatch): + """THE behavioural test for the ContextVar fix. With the old + process-global counter, the second CONCURRENT delegation was refused + ("max delegation depth reached" — observed live, two sibling calls + 1.8ms apart); with a per-context depth both proceed, while a NESTED + delegation inside a subagent is still bounded at depth 1. + + `run_typed` and `get_subagent` are patched at their import site + (`braindb.agent.agent` — the tool imports them locally per call), so no + LLM and no DB are touched. + """ + from braindb.agent.schemas import SubagentResult + + state = {"active": 0, "max_active": 0, "nested_reply": None} + + async def call_delegate(task: str) -> str: + return await tools_mod.delegate_to_subagent.on_invoke_tool( + _Ctx(), json.dumps({"task": task})) + + async def fake_run_typed(task, agent, schema, max_turns=None): + state["active"] += 1 + state["max_active"] = max(state["max_active"], state["active"]) + await asyncio.sleep(0.05) # hold the slot so the two runs overlap + if state["nested_reply"] is None: + # A subagent trying to delegate further must be refused — + # this runs inside the caller's depth-1 context. + state["nested_reply"] = await call_delegate("nested probe") + state["active"] -= 1 + return SubagentResult(result=f"ok:{task}") + + monkeypatch.setattr(agent_mod, "run_typed", fake_run_typed) + monkeypatch.setattr(agent_mod, "get_subagent", lambda: object()) + + async def main(): + return await asyncio.gather(call_delegate("A"), call_delegate("B")) + + r1, r2 = asyncio.run(main()) + assert "ok:A" in r1, r1 + assert "ok:B" in r2, r2 + # Genuine overlap — a global counter would have refused the second call + # and max_active would never reach 2. + assert state["max_active"] == 2 + assert "max delegation depth reached" in (state["nested_reply"] or "") + + +def test_delegation_depth_defaults_to_zero_and_resets(): + assert tools_mod._depth_var.get() == 0 + + +def test_max_depth_still_one(): + """Bounded delegation is deliberate — the fix was the scope of the + counter, not the limit.""" + assert tools_mod._MAX_DEPTH == 1 + + +# ====================================================================== # +# Run tag — concurrent runs must be separable in the logs # +# ====================================================================== # + +def test_verbose_tool_lines_carry_the_current_run_tag(monkeypatch, caplog): + """Audits of concurrent writer+maintainer+subagent logs previously had + to attribute TOOL lines by argument fingerprint. `run_typed` sets a + per-run tag; `_verbose` must print it.""" + import logging as _logging + from braindb.agent.run_state import reset_run_tag, set_run_tag + from braindb.agent.schemas import SubagentResult + + monkeypatch.setattr(tools_mod.settings, "agent_verbose", True) + + async def fake_run_typed(task, agent, schema, max_turns=None): + return SubagentResult(result="ok") + + monkeypatch.setattr(agent_mod, "run_typed", fake_run_typed) + monkeypatch.setattr(agent_mod, "get_subagent", lambda: object()) + + token = set_run_tag("tag4242") + try: + with caplog.at_level(_logging.INFO, logger="braindb.agent.tools"): + reply = asyncio.run( + tools_mod.delegate_to_subagent.on_invoke_tool( + _Ctx(), json.dumps({"task": "probe"}))) + assert "ok" in reply + tool_lines = [r.message for r in caplog.records + if "delegate_to_subagent" in r.getMessage()] + assert tool_lines, "no TOOL lines captured" + assert any("[tag4242]" in r.getMessage() for r in caplog.records) + finally: + reset_run_tag(token) + + +# ====================================================================== # +# Reasoning effort — wiki-scoped, never the general agent # +# ====================================================================== # + +def test_reasoning_effort_default_is_a_no_op(): + """Blank default must send nothing, so behaviour is unchanged until an + operator opts in.""" + from braindb.config import settings + assert settings.agent_wiki_reasoning_effort == "" + agent_mod._cache.clear() + try: + for factory in (agent_mod.get_agent, agent_mod.get_maintainer_agent, + agent_mod.get_writer_agent, agent_mod.get_subagent): + ms = factory().model_settings + assert ms.extra_body is None + assert "reasoning_effort" not in ms.extra_args + finally: + agent_mod._cache.clear() + + +def test_reasoning_effort_never_travels_as_a_plain_param(): + """THE regression guard for a live outage. The SDK lifts a top-level + `reasoning_effort` out of extra_args/extra_body and promotes it to a + kwarg on litellm.acompletion(), where the `openai` provider allow-list + rejects it — 59 triage jobs died this way. It must ride nested inside + `chat_template_kwargs`, which nothing intercepts or filters.""" + from braindb.config import settings + monkey = settings.agent_wiki_reasoning_effort + try: + object.__setattr__(settings, "agent_wiki_reasoning_effort", "low") + except Exception: + settings.agent_wiki_reasoning_effort = "low" + agent_mod._cache.clear() + try: + ms = agent_mod.get_writer_agent().model_settings + assert ms.extra_body == {"chat_template_kwargs": + {"reasoning_effort": "low"}} + # the key the SDK intercepts must appear at NEITHER top level + assert "reasoning_effort" not in ms.extra_body + assert "reasoning_effort" not in ms.extra_args + finally: + settings.agent_wiki_reasoning_effort = monkey + agent_mod._cache.clear() + + +def test_reasoning_effort_reaches_wiki_agents_but_not_the_general_agent(monkeypatch): + """THE scoping guarantee. `get_agent()` is shared by /agent/query AND the + ingest watcher, whose extraction runs are the most reasoning-dependent + work in the stack — it must never inherit the wiki setting. This test + fails if anyone later threads it there.""" + from braindb.config import settings + monkeypatch.setattr(settings, "agent_wiki_reasoning_effort", "low") + agent_mod._cache.clear() + try: + for factory in (agent_mod.get_maintainer_agent, + agent_mod.get_writer_agent, agent_mod.get_subagent): + ms = factory().model_settings + assert ms.extra_body == {"chat_template_kwargs": + {"reasoning_effort": "low"}}, factory.__name__ + # the transport deadline must survive alongside it + assert ms.extra_args.get("timeout") == settings.agent_request_timeout + assert agent_mod.get_agent().model_settings.extra_body is None + finally: + agent_mod._cache.clear()