Compare commits

..
Author SHA1 Message Date
mathiasandClaude Opus 5.5 22d0653b61 test(watcher): a marker link planted during the run is refused and not retried (#158)
CI / Lint / Test / Vet (pull_request) Successful in 1m3s
CI / Mirror to GitHub (pull_request) Skipped
Review round 2 on PR #195: the marker writes had no test, so reverting
them to by-path writes, or the refusal to log-only, stayed green. The
link is planted from inside the fake model call, between the walk's
Lstat and the marker write, for both markers and both an outside and
an in-brain relative target; red on all four flips. The marker-entry
skip is now checked for .failed too.

Also pins cleanRel's acceptance of a backslash as a filename byte in
brainfs itself, and stops the buildAndWrite control claiming writes it
does not count.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 23:47:24 +02:00
mathiasandClaude Opus 5.5 1a2a93d2f4 fix(watcher): refusals end in a terminal state; pin every converted writer (#158)
CI / Lint / Test / Vet (pull_request) Successful in 1m6s
CI / Mirror to GitHub (pull_request) Skipped
Review round 1 on PR #195:
- A refused marker or archive write left no marker, so the note was
  reprocessed every interval (archive copies piling up; an LLM call per
  round on the failure path). The marker check now uses Lstat (any entry
  at the marker path counts, so a dangling marker link skips the note
  instead of looping), and a note whose archive copy or marker write is
  refused is remembered (path, size, mtime) and not retried until it
  changes.
- cleanRel no longer refuses a backslash: it is a legal filename byte on
  Linux, not a traversal, and a raw note named with one was never
  archived or marked.
- Tests now reach every converted writer: the log.md append (real
  raw/failed so the log is reached, symlinked log.md), a dangling marker
  link, the refused-retry loop with a positive control, a backslash name,
  a buildAndWrite swap race, and the claude-session archive with in-brain
  and outside symlinked archive dirs. The stale lint allowlist entry for
  appendWatcherLog is removed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 23:40:19 +02:00
mathiasandClaude Opus 5.5 77c44f298c fix: pipeline, backfill, watcher and archive writes go through the brain root (#158 part B)
CI / Lint / Test / Vet (pull_request) Successful in 1m13s
CI / Mirror to GitHub (pull_request) Skipped
The remaining writers created or rewrote brain files by path:
- pipeline buildAndWrite: os.MkdirAll + os.WriteFile of every page.
- BackfillRefs: os.WriteFile of each concept page; a page swapped for a
  symlink after the read was written through (the race test overwrote an
  in-brain target before this change).
- watcher: the raw/processed and raw/failed archive copies (os.Create),
  the .processed/.failed markers and the log.md entry (os.OpenFile).
- cmd/server: the claude-session archive.

All now go through brainfs PutRegular/AppendRegular (new package-level
PutFile/AppendFile for one-off writers: the watcher's call-site lint bans
.Open calls in that file so raw/ reads stay on OpenRegular).

Stacked on #189 (needs PutRegular/AppendRegular).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 23:26:02 +02:00
48 changed files with 507 additions and 3163 deletions
-12
View File
@@ -84,15 +84,3 @@ When acting as a coding agent on this project:
4. Never modify files outside the project root without explicit permission 4. Never modify files outside the project root without explicit permission
5. When adding a dependency, explain why in the commit message 5. When adding a dependency, explain why in the commit message
6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM 6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM
## `main` is protected: ship by PR (local-dev#42, decided 2026-10-08)
Gitea refuses direct pushes to `main` in this repo, and a PR merges only after its CI status (`CI / Lint / Test / Vet (pull_request)`) passes.
**Why:** local-dev rule 5's reviewed lane (risk/medium+ is reviewed by a human) was enforced on the merge tools
(`tbd_ship`, `pr_merge`: gitea-mcp v0.17.0/v0.18.0, merge only for an issue labelled exactly `risk/low` with green CI)
but not on a direct commit to `main`, which bypassed it. This overrides rule 5's "risk/low commits straight to main"
for this repo; control code is where the lane matters.
**How:** `tbd_ship` for a one-file change, or branch -> PR -> `pr_merge` (name the `risk/low` issue); above `risk/low`
open the PR and a human merges it in the Gitea UI after review. Tags are not protected. CD jobs push image tags to
the **infra** repo with deploy keys, never to this repo's `main`. Undo (Mathias only): Settings -> Branches, or
`PATCH /repos/mathias/hyperguild/branch_protections/main {"enable_push": true}`.
+17 -81
View File
@@ -164,63 +164,6 @@ These rules apply to every task across every project, regardless of harness.
the thing it was named for was exploitable. If you cannot name the mechanism, the thing it was named for was exploitable. If you cannot name the mechanism,
you do not yet know what the test observes. you do not yet know what the test observes.
10. **Short sessions, one rc per repo, edit only in a locked worktree of the repo
you are changing** (ADR-0002, accepted 2026-10-06).
- One session per issue or task, ending with rule 6's close ritual. Related
issues may share a session only when they share a worktree and an owner.
State that must outlive the session goes in an issue, not the chat.
- Start `claude rc` in the repo you will edit (`infra`, `trucker-ecommerce`,
`hyperguild`, ...), not in `~/dev`: the rc worktree is a worktree of the
launch directory's repo, so launched from `~/dev` it protects nothing you
edit (measured: 73% of edits landed in shared checkouts). Start it only
while working there.
- Keep at most one thin `~/dev` coordinator session. It files issues and hands
out tasks; it reads and edits no sub-repo code.
- Split cross-repo work per repo. The contract between the halves lives in an
issue on each repo.
- Edit in a worktree of that repo, locked at creation:
`scripts/wt.sh add <repo> <task>` does it (fetch, `wt/<task>` from origin/main, lock);
`scripts/wt.sh done <repo>-<task>` removes it after the push and refuses dirty or
unpushed work. By hand: `git worktree add ~/worktrees/<name> -b <branch> origin/main`
then `git worktree lock --reason "manual: <purpose>" ~/worktrees/<name>`.
- Before working a repo another session may touch, announce your issue list to
that session (`SendMessage`) and ask for theirs.
- Auto-memory is keyed by cwd, so a repo rc starts without it. Cross-repo
working rules live in brain (`wiki/homelab/decisions/2026-10-06-062416-auto.md`, wing homelab, hall
decisions) and the section below, not in per-directory memory.
## Standing footguns
Each cost real time once. Detail and evidence: brain `wiki/homelab/decisions/2026-10-06-062416-auto.md` (title standing-working-rules).
- **Gate then commit with `&&`, never `;`.** "Green" means exit code 0 and
`grep -c FAIL` is 0, not a filtered grep.
- **Stage by path, never `git add -A` or `git add .`** (they stage from the repo
root and swept in other sessions' work twice). Check `git status --porcelain` first.
- **Before committing in a shared checkout:** `git branch --show-current`.
- **Before tagging:** `git fetch origin --tags`, then `git describe --tags --abbrev=0 <sha>`
and `git tag --sort=-v:refname | head`; go above both. Never take the last tag
from a report.
- **Run only what the project's gate runs.** No repo-wide formatter unless the gate
runs it.
- **Reproduce a CI red at the cited commit before applying an issue's prescribed fix.**
- **Read an issue's comments (`issue_list_comments`), not just its body, before working it.**
- **A comment naming a control ("behind branch protection") is not evidence; query
the control.** A working detector can hide a missing preventer.
- **On koala the MCP servers are `mcp__plugin_skills_{brain,gitea,infra}__*`.** The
per-turn auth reminder is not evidence; call the tool or run `claude mcp list`.
Brain writes always pass `wing` + `hall`.
- **`cat /etc/hostname` before assuming which host you are on.**
- **Never run bare `git remote -v` or `git config --list`** (url.insteadOf expands
into a live token); use `git config --get remote.origin.url`.
- **On koala `sudo` is denied outright; k8s exec, env dumps and flux reconcile need
per-action approval; decrypting a SOPS file and printing in one Bash call is
hook-blocked.** Do the repo-side work and hand the host-side step over in an issue.
- **`set -o pipefail` plus an optional `grep | head | awk` dies on no match:** append `|| true`.
- **Gitea `file_write_branch` takes plain text, not base64.**
- **`OP_SERVICE_ACCOUNT_TOKEN` in the shell is read-only for `op`.** Operational
secrets go in the AgentSecrets vault via `set -a; . ~/.op-agent-write.env; set +a`.
## CAD sprint patterns ## CAD sprint patterns
Two CAD sprint archetypes (defs: brain `wiki/homelab/decisions/continuous-agentic-development-cad-concept-2026-06-16`): Two CAD sprint archetypes (defs: brain `wiki/homelab/decisions/continuous-agentic-development-cad-concept-2026-06-16`):
@@ -301,20 +244,25 @@ Three machines on Tailscale:
All development repos live at `~/dev/` (softlink from `~/Documents/local-dev/`). All development repos live at `~/dev/` (softlink from `~/Documents/local-dev/`).
Organized by **role**, because the role decides which rules apply (ADR-0003): Organized in thematic folders:
| Folder | Role | Rules that key off it | | Folder | Focus | Count |
|--------|------|-----------------------| |--------|-------|-------|
| `platform/` | infra, brain, hyperguild, gitea-mcp, dispatch, shared tooling | control code (rule 8), GitOps review lane | | `GO/` | Go web frameworks, API integrations, learning projects | ~10 |
| `products/` | tapir, cobalt-dingo, openbanking-api, ... | ordinary trunk-based work | | `AI/` | ML research, AI frameworks (FinRL, DSPy, crawl4ai) | ~6 |
| `clients/` | client work | local models only, never cloud APIs (Client work rules) | | `AGENTS/` | Autonomous agents, coding agents, MCP servers, infra | ~15 |
| `spikes/` | throwaway | no ceremony | | `QKX/` | Invoice processing, financial automation, payment systems | ~13 |
| `XT/` | Climate data, sustainability (Klimatkollen, Garbo) | ~2 |
New projects are born into a role folder (`task new-project -- <name> <client|personal> See `~/dev/PROJECT_SUMMARY.md` for detailed descriptions of each project.
<role-folder>`); a client tag requires `clients/` and `clients/` requires a client tag.
Existing repos still sit in the old `AGENTS/` and `AI/` folders until they are moved one by ### Key active projects
one (ADR-0003 step 3). `~/dev/PROJECT_SUMMARY.md` (generated) lists every repo and where it
is. How a session runs on koala (worktrees, rc per repo, the reaper, memory): `docs/working-on-koala.md`. - **super-koala** (`AGENTS/`) — multi-component agent stack with LangGraph, DSPy, MCP
- **azure-tiger** (`QKX/`) — invoice extraction → ISO 20022 payment instructions
- **gocrwl** (`AGENTS/`) — Go web crawler with containerized deployment
- **koala-ai-stack** (`AGENTS/`) — local AI server infrastructure management
- **klimatkollen** (`XT/`) — Swedish municipal climate data platform
## Knowledge base — actively use it ## Knowledge base — actively use it
@@ -566,16 +514,4 @@ When acting as a coding agent on this project:
5. When adding a dependency, explain why in the commit message 5. When adding a dependency, explain why in the commit message
6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM 6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM
## `main` is protected: ship by PR (local-dev#42, decided 2026-10-08)
Gitea refuses direct pushes to `main` in this repo, and a PR merges only after its CI status (`CI / Lint / Test / Vet (pull_request)`) passes.
**Why:** local-dev rule 5's reviewed lane (risk/medium+ is reviewed by a human) was enforced on the merge tools
(`tbd_ship`, `pr_merge`: gitea-mcp v0.17.0/v0.18.0, merge only for an issue labelled exactly `risk/low` with green CI)
but not on a direct commit to `main`, which bypassed it. This overrides rule 5's "risk/low commits straight to main"
for this repo; control code is where the lane matters.
**How:** `tbd_ship` for a one-file change, or branch -> PR -> `pr_merge` (name the `risk/low` issue); above `risk/low`
open the PR and a human merges it in the Gitea UI after review. Tags are not protected. CD jobs push image tags to
the **infra** repo with deploy keys, never to this repo's `main`. Undo (Mathias only): Settings -> Branches, or
`PATCH /repos/mathias/hyperguild/branch_protections/main {"enable_push": true}`.
--- ---
+17 -81
View File
@@ -159,63 +159,6 @@ These rules apply to every task across every project, regardless of harness.
the thing it was named for was exploitable. If you cannot name the mechanism, the thing it was named for was exploitable. If you cannot name the mechanism,
you do not yet know what the test observes. you do not yet know what the test observes.
10. **Short sessions, one rc per repo, edit only in a locked worktree of the repo
you are changing** (ADR-0002, accepted 2026-10-06).
- One session per issue or task, ending with rule 6's close ritual. Related
issues may share a session only when they share a worktree and an owner.
State that must outlive the session goes in an issue, not the chat.
- Start `claude rc` in the repo you will edit (`infra`, `trucker-ecommerce`,
`hyperguild`, ...), not in `~/dev`: the rc worktree is a worktree of the
launch directory's repo, so launched from `~/dev` it protects nothing you
edit (measured: 73% of edits landed in shared checkouts). Start it only
while working there.
- Keep at most one thin `~/dev` coordinator session. It files issues and hands
out tasks; it reads and edits no sub-repo code.
- Split cross-repo work per repo. The contract between the halves lives in an
issue on each repo.
- Edit in a worktree of that repo, locked at creation:
`scripts/wt.sh add <repo> <task>` does it (fetch, `wt/<task>` from origin/main, lock);
`scripts/wt.sh done <repo>-<task>` removes it after the push and refuses dirty or
unpushed work. By hand: `git worktree add ~/worktrees/<name> -b <branch> origin/main`
then `git worktree lock --reason "manual: <purpose>" ~/worktrees/<name>`.
- Before working a repo another session may touch, announce your issue list to
that session (`SendMessage`) and ask for theirs.
- Auto-memory is keyed by cwd, so a repo rc starts without it. Cross-repo
working rules live in brain (`wiki/homelab/decisions/2026-10-06-062416-auto.md`, wing homelab, hall
decisions) and the section below, not in per-directory memory.
## Standing footguns
Each cost real time once. Detail and evidence: brain `wiki/homelab/decisions/2026-10-06-062416-auto.md` (title standing-working-rules).
- **Gate then commit with `&&`, never `;`.** "Green" means exit code 0 and
`grep -c FAIL` is 0, not a filtered grep.
- **Stage by path, never `git add -A` or `git add .`** (they stage from the repo
root and swept in other sessions' work twice). Check `git status --porcelain` first.
- **Before committing in a shared checkout:** `git branch --show-current`.
- **Before tagging:** `git fetch origin --tags`, then `git describe --tags --abbrev=0 <sha>`
and `git tag --sort=-v:refname | head`; go above both. Never take the last tag
from a report.
- **Run only what the project's gate runs.** No repo-wide formatter unless the gate
runs it.
- **Reproduce a CI red at the cited commit before applying an issue's prescribed fix.**
- **Read an issue's comments (`issue_list_comments`), not just its body, before working it.**
- **A comment naming a control ("behind branch protection") is not evidence; query
the control.** A working detector can hide a missing preventer.
- **On koala the MCP servers are `mcp__plugin_skills_{brain,gitea,infra}__*`.** The
per-turn auth reminder is not evidence; call the tool or run `claude mcp list`.
Brain writes always pass `wing` + `hall`.
- **`cat /etc/hostname` before assuming which host you are on.**
- **Never run bare `git remote -v` or `git config --list`** (url.insteadOf expands
into a live token); use `git config --get remote.origin.url`.
- **On koala `sudo` is denied outright; k8s exec, env dumps and flux reconcile need
per-action approval; decrypting a SOPS file and printing in one Bash call is
hook-blocked.** Do the repo-side work and hand the host-side step over in an issue.
- **`set -o pipefail` plus an optional `grep | head | awk` dies on no match:** append `|| true`.
- **Gitea `file_write_branch` takes plain text, not base64.**
- **`OP_SERVICE_ACCOUNT_TOKEN` in the shell is read-only for `op`.** Operational
secrets go in the AgentSecrets vault via `set -a; . ~/.op-agent-write.env; set +a`.
## CAD sprint patterns ## CAD sprint patterns
Two CAD sprint archetypes (defs: brain `wiki/homelab/decisions/continuous-agentic-development-cad-concept-2026-06-16`): Two CAD sprint archetypes (defs: brain `wiki/homelab/decisions/continuous-agentic-development-cad-concept-2026-06-16`):
@@ -296,20 +239,25 @@ Three machines on Tailscale:
All development repos live at `~/dev/` (softlink from `~/Documents/local-dev/`). All development repos live at `~/dev/` (softlink from `~/Documents/local-dev/`).
Organized by **role**, because the role decides which rules apply (ADR-0003): Organized in thematic folders:
| Folder | Role | Rules that key off it | | Folder | Focus | Count |
|--------|------|-----------------------| |--------|-------|-------|
| `platform/` | infra, brain, hyperguild, gitea-mcp, dispatch, shared tooling | control code (rule 8), GitOps review lane | | `GO/` | Go web frameworks, API integrations, learning projects | ~10 |
| `products/` | tapir, cobalt-dingo, openbanking-api, ... | ordinary trunk-based work | | `AI/` | ML research, AI frameworks (FinRL, DSPy, crawl4ai) | ~6 |
| `clients/` | client work | local models only, never cloud APIs (Client work rules) | | `AGENTS/` | Autonomous agents, coding agents, MCP servers, infra | ~15 |
| `spikes/` | throwaway | no ceremony | | `QKX/` | Invoice processing, financial automation, payment systems | ~13 |
| `XT/` | Climate data, sustainability (Klimatkollen, Garbo) | ~2 |
New projects are born into a role folder (`task new-project -- <name> <client|personal> See `~/dev/PROJECT_SUMMARY.md` for detailed descriptions of each project.
<role-folder>`); a client tag requires `clients/` and `clients/` requires a client tag.
Existing repos still sit in the old `AGENTS/` and `AI/` folders until they are moved one by ### Key active projects
one (ADR-0003 step 3). `~/dev/PROJECT_SUMMARY.md` (generated) lists every repo and where it
is. How a session runs on koala (worktrees, rc per repo, the reaper, memory): `docs/working-on-koala.md`. - **super-koala** (`AGENTS/`) — multi-component agent stack with LangGraph, DSPy, MCP
- **azure-tiger** (`QKX/`) — invoice extraction → ISO 20022 payment instructions
- **gocrwl** (`AGENTS/`) — Go web crawler with containerized deployment
- **koala-ai-stack** (`AGENTS/`) — local AI server infrastructure management
- **klimatkollen** (`XT/`) — Swedish municipal climate data platform
## Knowledge base — actively use it ## Knowledge base — actively use it
@@ -560,15 +508,3 @@ When acting as a coding agent on this project:
4. Never modify files outside the project root without explicit permission 4. Never modify files outside the project root without explicit permission
5. When adding a dependency, explain why in the commit message 5. When adding a dependency, explain why in the commit message
6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM 6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM
## `main` is protected: ship by PR (local-dev#42, decided 2026-10-08)
Gitea refuses direct pushes to `main` in this repo, and a PR merges only after its CI status (`CI / Lint / Test / Vet (pull_request)`) passes.
**Why:** local-dev rule 5's reviewed lane (risk/medium+ is reviewed by a human) was enforced on the merge tools
(`tbd_ship`, `pr_merge`: gitea-mcp v0.17.0/v0.18.0, merge only for an issue labelled exactly `risk/low` with green CI)
but not on a direct commit to `main`, which bypassed it. This overrides rule 5's "risk/low commits straight to main"
for this repo; control code is where the lane matters.
**How:** `tbd_ship` for a one-file change, or branch -> PR -> `pr_merge` (name the `risk/low` issue); above `risk/low`
open the PR and a human merges it in the Gitea UI after review. Tags are not protected. CD jobs push image tags to
the **infra** repo with deploy keys, never to this repo's `main`. Undo (Mathias only): Settings -> Branches, or
`PATCH /repos/mathias/hyperguild/branch_protections/main {"enable_push": true}`.
-12
View File
@@ -84,15 +84,3 @@ When acting as a coding agent on this project:
4. Never modify files outside the project root without explicit permission 4. Never modify files outside the project root without explicit permission
5. When adding a dependency, explain why in the commit message 5. When adding a dependency, explain why in the commit message
6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM 6. For client projects: never send code or context to cloud APIs — use local models via LiteLLM
## `main` is protected: ship by PR (local-dev#42, decided 2026-10-08)
Gitea refuses direct pushes to `main` in this repo, and a PR merges only after its CI status (`CI / Lint / Test / Vet (pull_request)`) passes.
**Why:** local-dev rule 5's reviewed lane (risk/medium+ is reviewed by a human) was enforced on the merge tools
(`tbd_ship`, `pr_merge`: gitea-mcp v0.17.0/v0.18.0, merge only for an issue labelled exactly `risk/low` with green CI)
but not on a direct commit to `main`, which bypassed it. This overrides rule 5's "risk/low commits straight to main"
for this repo; control code is where the lane matters.
**How:** `tbd_ship` for a one-file change, or branch -> PR -> `pr_merge` (name the `risk/low` issue); above `risk/low`
open the PR and a human merges it in the Gitea UI after review. Tags are not protected. CD jobs push image tags to
the **infra** repo with deploy keys, never to this repo's `main`. Undo (Mathias only): Settings -> Branches, or
`PATCH /repos/mathias/hyperguild/branch_protections/main {"enable_push": true}`.
+15 -41
View File
@@ -17,6 +17,7 @@ import (
"github.com/mathiasbq/hyperguild/ingestion/internal/api" "github.com/mathiasbq/hyperguild/ingestion/internal/api"
"github.com/mathiasbq/hyperguild/ingestion/internal/audit" "github.com/mathiasbq/hyperguild/ingestion/internal/audit"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/capture" "github.com/mathiasbq/hyperguild/ingestion/internal/capture"
"github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp" "github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp"
"github.com/mathiasbq/hyperguild/ingestion/internal/classification" "github.com/mathiasbq/hyperguild/ingestion/internal/classification"
@@ -36,7 +37,6 @@ import (
"github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore" "github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore"
"github.com/mathiasbq/hyperguild/ingestion/internal/watcher" "github.com/mathiasbq/hyperguild/ingestion/internal/watcher"
"github.com/mathiasbq/hyperguild/ingestion/internal/webhook" "github.com/mathiasbq/hyperguild/ingestion/internal/webhook"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
"k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest" "k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd" "k8s.io/client-go/tools/clientcmd"
@@ -90,11 +90,15 @@ func (s *claudeSink) Ingest(ctx context.Context, b claudewatcher.Batch) error {
sb.WriteString("\n\n") sb.WriteString("\n\n")
} }
slug := "session-" + b.Host + "-" + b.SessionID slug := "session-" + b.Host + "-" + b.SessionID
dest := filepath.Join(s.brainDir, "archive", "claude-sessions", b.Host, slug+".md") dest := "archive/claude-sessions/" + b.Host + "/" + slug + ".md"
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil { // Written through the brain root, never through a symlinked directory
return fmt.Errorf("create claude-sessions archive dir: %w", err) // or file (hyperguild#158).
bfs, err := brainfs.Open(s.brainDir)
if err != nil {
return fmt.Errorf("open brain: %w", err)
} }
if err := os.WriteFile(dest, []byte(sb.String()), 0o644); err != nil { defer func() { _ = bfs.Close() }()
if err := bfs.PutRegular(dest, []byte(sb.String())); err != nil {
return fmt.Errorf("write claude session archive: %w", err) return fmt.Errorf("write claude session archive: %w", err)
} }
s.logger.Debug("claude session archived (non-indexed)", "path", dest) s.logger.Debug("claude session archived (non-indexed)", "path", dest)
@@ -145,17 +149,6 @@ func envInt(key string, fallback int) int {
return fallback return fallback
} }
// envFloat reads a confidence-style value in [0,1]. Anything unparseable or
// outside the range falls back: a threshold of 0 would auto-place everything.
func envFloat(key string, fallback float64) float64 {
if v := os.Getenv(key); v != "" {
if f, err := strconv.ParseFloat(v, 64); err == nil && f >= 0 && f <= 1 {
return f
}
}
return fallback
}
// buildNotifier returns the ntfy notifier, or a nil interface when // buildNotifier returns the ntfy notifier, or a nil interface when
// BRAIN_NTFY_URL is unset. A nil *interface* (not a typed-nil) is required so // BRAIN_NTFY_URL is unset. A nil *interface* (not a typed-nil) is required so
// DegradingSink/Reconcile and the governance observer skip it cleanly. // DegradingSink/Reconcile and the governance observer skip it cleanly.
@@ -191,12 +184,9 @@ func alertingWarnings(getenv func(string) string) []string {
return out return out
} }
// buildNotifier returns the ntfy notifier for one alert source, titled so func buildNotifier() audit.Notifier {
// the source is recognisable (hyperguild#163), or nil when ntfy is not if n := audit.NewNtfyNotifier(os.Getenv("BRAIN_NTFY_URL"), os.Getenv("BRAIN_NTFY_TOKEN")); n != nil {
// configured. return n
func buildNotifier(getenv func(string) string, title string) audit.Notifier {
if n := audit.NewNtfyNotifier(getenv("BRAIN_NTFY_URL"), getenv("BRAIN_NTFY_TOKEN")); n != nil {
return n.WithTitle(title)
} }
return nil return nil
} }
@@ -218,7 +208,7 @@ func buildAuditSink(ctx context.Context, brainDir string, logger *slog.Logger) c
logger.Error("capture audit buffer init", "err", err) logger.Error("capture audit buffer init", "err", err)
os.Exit(1) os.Exit(1)
} }
notifier := buildNotifier(os.Getenv, "brain-capture audit degraded") notifier := buildNotifier()
reconcileInterval := time.Duration(envInt("BRAIN_AUDIT_RECONCILE_INTERVAL", 60)) * time.Second reconcileInterval := time.Duration(envInt("BRAIN_AUDIT_RECONCILE_INTERVAL", 60)) * time.Second
audit.StartReconcile(ctx, central, buffer, notifier, reconcileInterval) audit.StartReconcile(ctx, central, buffer, notifier, reconcileInterval)
logger.Info("capture audit: loki+buffer sink", "loki", lokiURL, "reconcile_s", int(reconcileInterval.Seconds())) logger.Info("capture audit: loki+buffer sink", "loki", lokiURL, "reconcile_s", int(reconcileInterval.Seconds()))
@@ -381,7 +371,7 @@ func main() {
ctx := context.Background() ctx := context.Background()
// hyperguild#157: report a symlink that appears in the brain while running. // hyperguild#157: report a symlink that appears in the brain while running.
go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier(os.Getenv, "brain symlink tripwire")) go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier())
// Never fatal on an unreachable issuer (#83): on a cold cluster boot this // Never fatal on an unreachable issuer (#83): on a cold cluster boot this
// pod can beat dex to serving and get a 502 from ingress, and exiting meant // pod can beat dex to serving and get a 502 from ingress, and exiting meant
@@ -401,26 +391,10 @@ func main() {
} }
if watchInterval > 0 { if watchInterval > 0 {
// INGEST_WING_RESOLVE=1 routes raw/ notes into wiki/<wing>/ by
// subfolder, frontmatter or classifier+scorer, and parks the rest in
// raw/needs-review/ (hyperguild#205). Off = the retired layout. Set
// explicitly, a failure to build the resolver is fatal: silently
// falling back would write into the layout this flag exists to leave.
var wingResolver *wing.Resolver
if os.Getenv("INGEST_WING_RESOLVE") == "1" {
var rerr error
wingResolver, rerr = wing.NewConfiguredResolver(brainDir, llmClient.Complete, envFloat("INGEST_WING_THRESHOLD", 0.9))
if rerr != nil {
logger.Error("build wing resolver", "err", rerr)
os.Exit(1)
}
logger.Info("wing resolver enabled", "wings", len(wingResolver.Known), "threshold", wingResolver.Threshold)
}
watcher.Start(ctx, watcher.Config{ watcher.Start(ctx, watcher.Config{
BrainDir: brainDir, BrainDir: brainDir,
Interval: time.Duration(watchInterval) * time.Second, Interval: time.Duration(watchInterval) * time.Second,
Pipeline: pipelineCfg, Pipeline: pipelineCfg,
Resolver: wingResolver,
}) })
} }
@@ -577,7 +551,7 @@ func main() {
// capture already gets, plus an out-of-band alert. Wired here because // capture already gets, plus an out-of-band alert. Wired here because
// brain_edit_root needs the capture plumbing anyway — that is what // brain_edit_root needs the capture plumbing anyway — that is what
// re-derives the static principal its gate checks. // re-derives the static principal its gate checks.
mcpSrv.WithGovernanceObserver(auditSink, buildNotifier(os.Getenv, "brain governance edit")) mcpSrv.WithGovernanceObserver(auditSink, buildNotifier())
logger.Info("capture enabled (REST + MCP tool)", "sovereign_principals", len(sovereign), logger.Info("capture enabled (REST + MCP tool)", "sovereign_principals", len(sovereign),
"governance_observer", true) "governance_observer", true)
} else { } else {
+20 -16
View File
@@ -77,22 +77,26 @@ func TestAlertingWarnings(t *testing.T) {
} }
} }
func TestEnvFloat(t *testing.T) { // hyperguild#158: the claude-session archive was written with os.MkdirAll +
tests := []struct { // os.WriteFile by path, so a symlinked archive directory redirected it.
name string func TestClaudeSinkDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) {
set string for name, mk := range map[string]func(t *testing.T, dir string) (link, target string){
want float64 "outside": func(t *testing.T, _ string) (string, string) { o := t.TempDir(); return o, o },
}{ "in-brain": func(t *testing.T, dir string) (string, string) {
{"unset falls back", "", 0.9}, require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw"), 0o755))
{"valid value", "0.75", 0.75}, return filepath.Join("..", "raw"), filepath.Join(dir, "raw")
{"garbage falls back rather than becoming 0", "high", 0.9}, },
{"out of range is rejected, not clamped silently into auto-placing everything", "-1", 0.9}, } {
{"above one is rejected", "1.5", 0.9}, t.Run(name, func(t *testing.T) {
} dir := t.TempDir()
for _, tt := range tests { link, target := mk(t, dir)
t.Run(tt.name, func(t *testing.T) { require.NoError(t, os.Symlink(link, filepath.Join(dir, "archive")))
t.Setenv("TEST_ENV_FLOAT", tt.set) sink := &claudeSink{brainDir: dir, logger: slog.New(slog.NewTextHandler(os.Stderr, nil))}
assert.InDelta(t, tt.want, envFloat("TEST_ENV_FLOAT", 0.9), 1e-9) err := sink.Ingest(context.Background(), claudewatcher.Batch{Host: "koala", SessionID: "abc",
Turns: []claudewatcher.Turn{{Type: "assistant", Content: "x"}}})
assert.Error(t, err)
ents, _ := os.ReadDir(target)
assert.Empty(t, ents, "the archive was written at the link's target")
}) })
} }
} }
-13
View File
@@ -159,16 +159,3 @@ func TestBuildMuxObservesRequestsAndServesMetrics(t *testing.T) {
require.Equal(t, http.StatusOK, code) require.Equal(t, http.StatusOK, code)
assert.Contains(t, body, `path="GET /pending"`) assert.Contains(t, body, `path="GET /pending"`)
} }
// hyperguild#185: an authenticated request to an unregistered path reached
// the inner REST mux, which left Pattern empty, and the metrics label fell
// back to the raw URL path: one Prometheus series per unique path any token
// holder sends. It is now labelled with a fixed catch-all.
func TestBuildMuxLabelsAnUnregisteredPathWithAFixedLabel(t *testing.T) {
r := testMux(t)
code, _ := do(t, r, "GET", "/zqrandom-185-segment", "Bearer "+routesTok)
require.Equal(t, http.StatusNotFound, code)
_, body := do(t, r, "GET", "/metrics", "")
assert.NotContains(t, body, "zqrandom-185-segment", "the raw path became a metrics label")
assert.Contains(t, body, `path="unmatched"`)
}
@@ -1,9 +1,6 @@
package main package main
import ( import (
"context"
"net/http"
"net/http/httptest"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
@@ -77,16 +74,3 @@ func TestStartupSymlinkCheckRefusesABrainWithAFIFO(t *testing.T) {
require.Error(t, err) require.Error(t, err)
assert.Contains(t, err.Error(), "classification.yaml") assert.Contains(t, err.Error(), "classification.yaml")
} }
// hyperguild#163: each alert source gets its own ntfy title.
func TestBuildNotifierUsesTheGivenTitle(t *testing.T) {
var got string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { got = r.Header.Get("Title") }))
defer srv.Close()
env := map[string]string{"BRAIN_NTFY_URL": srv.URL}
n := buildNotifier(func(k string) string { return env[k] }, "brain symlink tripwire")
require.NotNil(t, n)
require.NoError(t, n.Notify(context.Background(), "m"))
assert.Equal(t, "brain symlink tripwire", got)
assert.Nil(t, buildNotifier(func(string) string { return "" }, "x"), "no ntfy URL, no notifier")
}
-147
View File
@@ -1,147 +0,0 @@
// Command wingbacktest measures how well the wing resolver would have placed
// pages whose wing a human already decided (hyperguild#205).
//
// It reads pages from the wings given in -wings, hides the wing, asks the
// local classifier and the BM25 scorer, and reports coverage/precision per
// confidence threshold. Page text is sent to -llm-url, which MUST be a local
// model (it is not checked here); sample only wings you would allow.
package main
import (
"context"
"flag"
"fmt"
"io/fs"
"log/slog"
"math/rand"
"os"
"path/filepath"
"sort"
"strings"
"time"
"github.com/mathiasbq/hyperguild/ingestion/internal/llm"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
func main() {
brain := flag.String("brain", "/var/lib/supervisor/brain", "brain directory (read only)")
wingsFlag := flag.String("wings", "", "comma-separated wings to sample pages FROM (must be wings you would allow)")
perWing := flag.Int("per-wing", 8, "pages sampled per wing")
llmURL := flag.String("llm-url", "http://10.0.1.20:31234/v1", "OpenAI-compatible LOCAL endpoint")
model := flag.String("model", "qwen35-9b-fast", "model name")
seed := flag.Int64("seed", 1, "sampling seed")
flag.Parse()
if err := run(*brain, strings.Split(*wingsFlag, ","), *perWing, *llmURL, *model, *seed); err != nil {
slog.Error("wingbacktest", "err", err)
os.Exit(1)
}
}
func run(brain string, sampleWings []string, perWing int, llmURL, model string, seed int64) error {
known, err := wing.KnownWings(brain)
if err != nil {
return err
}
client := llm.New(llmURL, "", model, 3*time.Minute)
classifier := wing.NewLLMClassifier(client.Complete)
rng := rand.New(rand.NewSource(seed))
var samples []wing.Sample
perWingSamples := map[string][]wing.Sample{}
for _, w := range sampleWings {
w = strings.TrimSpace(w)
if w == "" {
continue
}
pages, err := listPages(brain, w)
if err != nil {
return err
}
rng.Shuffle(len(pages), func(i, j int) { pages[i], pages[j] = pages[j], pages[i] })
if len(pages) > perWing {
pages = pages[:perWing]
}
for _, rel := range pages {
text, err := readBody(brain, rel)
if err != nil || len(text) < 200 {
continue
}
sugg, cerr := classifier.Classify(context.Background(), text, known)
if cerr != nil {
slog.Warn("classify failed", "page", rel, "err", cerr)
}
top, _ := wing.TopWing(wing.NewBM25Scorer(brain, known, rel).Score(text))
s := wing.Sample{Truth: w, Suggested: strings.ToLower(sugg.Wing), Confidence: sugg.Confidence, ScorerTop: top}
slog.Info("sample", "page", rel, "truth", w, "classifier", s.Suggested, "conf", s.Confidence, "scorer", top, "reason", sugg.Reason)
samples = append(samples, s)
perWingSamples[w] = append(perWingSamples[w], s)
}
}
report(samples, perWingSamples)
return nil
}
func report(samples []wing.Sample, per map[string][]wing.Sample) {
fmt.Printf("# wing backtest\n\nn=%d pages. Classifier+scorer must agree; threshold applies to classifier confidence.\n\n", len(samples))
fmt.Println("| threshold | auto | coverage | precision |")
fmt.Println("|---|---|---|---|")
for _, t := range []float64{0.5, 0.6, 0.7, 0.8, 0.9, 0.95} {
r := wing.Evaluate(samples, t)
fmt.Printf("| %.2f | %d/%d | %.0f%% | %.0f%% |\n", t, r.Auto, r.N, 100*r.Coverage, 100*r.Precision)
}
fmt.Println("\n## per wing (threshold 0.80)\n\n| wing | n | auto | precision |\n|---|---|---|---|")
names := make([]string, 0, len(per))
for w := range per {
names = append(names, w)
}
sort.Strings(names)
for _, w := range names {
r := wing.Evaluate(per[w], 0.8)
fmt.Printf("| %s | %d | %d | %.0f%% |\n", w, r.N, r.Auto, 100*r.Precision)
}
fmt.Println("\n## wrong auto-placements at 0.80")
fmt.Println()
for _, s := range samples {
if s.Suggested != "" && s.Confidence >= 0.8 && s.ScorerTop == s.Suggested && s.Suggested != s.Truth {
fmt.Printf("- truth=%s -> both said %s (conf %.2f)\n", s.Truth, s.Suggested, s.Confidence)
}
}
}
func listPages(brain, w string) ([]string, error) {
var out []string
root := filepath.Join(brain, "wiki", w)
err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
if err != nil || d.IsDir() || !d.Type().IsRegular() || !strings.HasSuffix(p, ".md") || d.Name() == "_index.md" {
return err
}
rel, rerr := filepath.Rel(brain, p)
if rerr != nil {
return rerr
}
out = append(out, filepath.ToSlash(rel))
return nil
})
if err != nil {
return nil, fmt.Errorf("walk %s: %w", w, err)
}
return out, nil
}
// readBody returns the page text without YAML frontmatter, so the wing label
// (and anything else in frontmatter) cannot leak into the classification.
func readBody(brain, rel string) (string, error) {
b, err := os.ReadFile(filepath.Join(brain, filepath.FromSlash(rel)))
if err != nil {
return "", fmt.Errorf("read %s: %w", rel, err)
}
s := string(b)
if strings.HasPrefix(s, "---\n") {
if end := strings.Index(s[4:], "\n---"); end >= 0 {
s = s[4+end+4:]
}
}
return strings.TrimSpace(s), nil
}
-92
View File
@@ -1,92 +0,0 @@
// Command wingmigrate moves pages out of the retired wiki/{sources,concepts,
// entities}/ layout into wiki/<wing>/ (hyperguild#205 S5).
//
// Default is a dry run that prints the plan. -apply performs the Auto items.
// Page text goes to -llm-url, which MUST be a local model.
package main
import (
"context"
"flag"
"fmt"
"log/slog"
"os"
"strings"
"time"
"github.com/mathiasbq/hyperguild/ingestion/internal/llm"
"github.com/mathiasbq/hyperguild/ingestion/internal/migrate"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
func main() {
brain := flag.String("brain", "/var/lib/supervisor/brain", "brain directory")
llmURL := flag.String("llm-url", "http://10.0.1.20:31234/v1", "OpenAI-compatible LOCAL endpoint")
model := flag.String("model", "qwen35-9b-fast", "model name")
threshold := flag.Float64("threshold", 0.9, "classifier confidence threshold")
apply := flag.Bool("apply", false, "perform the Auto moves (default: dry run)")
what := flag.String("what-if-eligible", "", "dry run only: comma-separated wings to treat as eligible, to see what tagging them would allow")
flag.Parse()
if *apply && *what != "" {
slog.Error("-what-if-eligible is a dry-run aid and cannot be combined with -apply: eligibility is a sovereignty decision made in classification.yaml")
os.Exit(2)
}
if err := run(*brain, *llmURL, *model, *threshold, *apply, *what); err != nil {
slog.Error("wingmigrate", "err", err)
os.Exit(1)
}
}
func run(brain, llmURL, model string, threshold float64, apply bool, whatIf string) error {
client := llm.New(llmURL, "", model, 3*time.Minute)
base, err := wing.NewConfiguredResolver(brain, client.Complete, threshold)
if err != nil {
return err
}
if whatIf != "" {
allowed := map[string]bool{}
for _, w := range strings.Split(whatIf, ",") {
allowed[strings.TrimSpace(w)] = true
}
base.Eligible = func(w string) bool { return allowed[w] }
}
rf := func(exclude string) *wing.Resolver {
r := *base
r.Scorer = wing.NewBM25Scorer(brain, base.Known, exclude)
return &r
}
items, err := migrate.BuildPlan(context.Background(), brain, rf)
if err != nil {
return err
}
report(items)
if !apply {
fmt.Println("\n_dry run: nothing was changed_")
return nil
}
n, err := migrate.Apply(brain, items)
fmt.Printf("\napplied %d moves\n", n)
return err
}
func report(items []migrate.Item) {
auto, exc := 0, 0
for _, it := range items {
if it.Status == migrate.Auto {
auto++
} else {
exc++
}
}
fmt.Printf("# wing migration plan\n\n%d pages: %d auto, %d exceptions\n\n", len(items), auto, exc)
fmt.Println("| status | kind | from | to / proposal | how | why |")
fmt.Println("|---|---|---|---|---|---|")
for _, it := range items {
dest := it.To
if dest == "" {
dest = it.Wing
}
fmt.Printf("| %s | %s | %s | %s | %s | %s |\n", it.Status, it.Kind, it.From, dest, it.Method, strings.ReplaceAll(it.Reason, "|", "/"))
}
}
@@ -1,51 +0,0 @@
package api
import (
"os"
"path/filepath"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func writeParked(t *testing.T, brainDir, name, content string) {
t.Helper()
dir := filepath.Join(brainDir, "raw", "needs-review")
require.NoError(t, os.MkdirAll(dir, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte(content), 0o644))
}
func TestListPending_ShowsParkedNotesWithTheirProposal(t *testing.T) {
dir := t.TempDir()
writeRaw(t, dir, "fresh.md", "---\ncreated_at: 2026-10-20T08:00:00Z\n---\nA note not yet processed.\n")
writeParked(t, dir, "unsure.md", "---\nproposed_wing: radar\nproposal_confidence: 0.55\nproposal_reason: \"below threshold\"\ncreated_at: 2026-10-19T08:00:00Z\n---\nBody of the unsure note.\n")
got, err := ListPending(dir)
require.NoError(t, err)
require.Len(t, got, 2)
assert.Equal(t, "unsure.md", got[0].Filename, "oldest first across both queues")
assert.Equal(t, "needs-review", got[0].Queue)
assert.Equal(t, "radar", got[0].ProposedWing)
assert.Equal(t, "below threshold", got[0].ProposalReason)
assert.InDelta(t, 0.55, got[0].ProposalConfidence, 1e-9)
assert.Contains(t, got[0].Excerpt, "Body of the unsure note")
assert.NotContains(t, got[0].Excerpt, "proposed_wing", "excerpt is body, not frontmatter")
assert.Equal(t, "fresh.md", got[1].Filename)
assert.Empty(t, got[1].Queue, "a top-level note has no queue label")
assert.Empty(t, got[1].ProposedWing)
}
func TestListPending_ParkedSymlinkIsNotListed(t *testing.T) {
dir := t.TempDir()
outside := filepath.Join(t.TempDir(), "secret.md")
require.NoError(t, os.WriteFile(outside, []byte("outside"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw", "needs-review"), 0o755))
require.NoError(t, os.Symlink(outside, filepath.Join(dir, "raw", "needs-review", "link.md")))
got, err := ListPending(dir)
require.NoError(t, err)
assert.Empty(t, got, "same symlink rule as the top level (#153)")
}
+4 -51
View File
@@ -8,7 +8,6 @@ import (
"path/filepath" "path/filepath"
"regexp" "regexp"
"sort" "sort"
"strconv"
"strings" "strings"
"time" "time"
@@ -21,15 +20,6 @@ type PendingNote struct {
CreatedAt string `json:"created_at"` CreatedAt string `json:"created_at"`
SizeBytes int64 `json:"size_bytes"` SizeBytes int64 `json:"size_bytes"`
Excerpt string `json:"excerpt"` Excerpt string `json:"excerpt"`
// Set only for a note parked in raw/needs-review/ by the wing resolver
// (hyperguild#205): it was not placed automatically, and the proposal says
// what the resolver would have done and why it did not. Approve by moving
// the note into raw/<wing>/ and the watcher ingests it there.
Queue string `json:"queue,omitempty"`
ProposedWing string `json:"proposed_wing,omitempty"`
ProposalReason string `json:"proposal_reason,omitempty"`
ProposalConfidence float64 `json:"proposal_confidence,omitempty"`
} }
// datePrefix matches a leading YYYY-MM-DD- on a raw filename, stripped when // datePrefix matches a leading YYYY-MM-DD- on a raw filename, stripped when
@@ -48,32 +38,11 @@ func ListPending(brainDir string) ([]PendingNote, error) {
return nil, fmt.Errorf("read raw dir: %w", err) return nil, fmt.Errorf("read raw dir: %w", err)
} }
defer func() { _ = root.Close() }() defer func() { _ = root.Close() }()
out, err := listNotes(root, "")
if err != nil {
return nil, err
}
// raw/needs-review/ is read through its own root, so a symlinked directory
// or note there is refused exactly like at the top level (#153).
if sub, serr := root.OpenRoot("needs-review"); serr == nil {
defer func() { _ = sub.Close() }()
parked, perr := listNotes(sub, "needs-review")
if perr != nil {
return nil, perr
}
out = append(out, parked...)
}
sort.SliceStable(out, func(i, j int) bool { return out[i].CreatedAt < out[j].CreatedAt })
return out, nil
}
// listNotes lists the regular .md notes directly under root. queue labels the
// notes ("" for the top of raw/) and enables reading a wing proposal.
func listNotes(root *os.Root, queue string) ([]PendingNote, error) {
entries, err := fs.ReadDir(root.FS(), ".") entries, err := fs.ReadDir(root.FS(), ".")
if err != nil { if err != nil {
return nil, fmt.Errorf("read raw dir: %w", err) return nil, fmt.Errorf("read raw dir: %w", err)
} }
out := make([]PendingNote, 0, len(entries)) out := make([]PendingNote, 0, len(entries))
for _, e := range entries { for _, e := range entries {
if !e.Type().IsRegular() || !strings.HasSuffix(e.Name(), ".md") || strings.HasPrefix(e.Name(), "tunnel-candidates-") { if !e.Type().IsRegular() || !strings.HasSuffix(e.Name(), ".md") || strings.HasPrefix(e.Name(), "tunnel-candidates-") {
@@ -88,20 +57,14 @@ func listNotes(root *os.Root, queue string) ([]PendingNote, error) {
if created == "" { if created == "" {
created = info.ModTime().UTC().Format(time.RFC3339) created = info.ModTime().UTC().Format(time.RFC3339)
} }
n := PendingNote{ out = append(out, PendingNote{
Filename: e.Name(), Filename: e.Name(),
CreatedAt: created, CreatedAt: created,
SizeBytes: info.Size(), SizeBytes: info.Size(),
Excerpt: excerpt(body, 200), Excerpt: excerpt(body, 200),
Queue: queue, })
}
if queue != "" {
n.ProposedWing = fm.get("proposed_wing")
n.ProposalReason = unquoteScalar(fm.get("proposal_reason"))
n.ProposalConfidence, _ = strconv.ParseFloat(fm.get("proposal_confidence"), 64)
}
out = append(out, n)
} }
sort.SliceStable(out, func(i, j int) bool { return out[i].CreatedAt < out[j].CreatedAt })
return out, nil return out, nil
} }
@@ -276,13 +239,3 @@ func readRawSource(brainDir, base string) ([]byte, error) {
} }
return raw, nil return raw, nil
} }
// unquoteScalar undoes the Go-style double quoting the watcher writes around a
// free-text value (strconv.Quote keeps it on one line); a bare value is
// returned as is.
func unquoteScalar(v string) string {
if u, err := strconv.Unquote(v); err == nil {
return u
}
return v
}
+1 -20
View File
@@ -13,14 +13,9 @@ import (
type NtfyNotifier struct { type NtfyNotifier struct {
topicURL string topicURL string
token string token string
title string
http *http.Client http *http.Client
} }
// defaultNtfyTitle is the title of a notifier built without WithTitle: the
// capture audit's degraded-state alert, its first user.
const defaultNtfyTitle = "brain-capture audit degraded"
// NewNtfyNotifier constructs a notifier for the given ntfy topic URL // NewNtfyNotifier constructs a notifier for the given ntfy topic URL
// (e.g. https://ntfy.sh/my-topic). token is an optional bearer for // (e.g. https://ntfy.sh/my-topic). token is an optional bearer for
// protected ntfy instances; it is held here and only sent in the // protected ntfy instances; it is held here and only sent in the
@@ -32,31 +27,17 @@ func NewNtfyNotifier(topicURL, token string) *NtfyNotifier {
return &NtfyNotifier{ return &NtfyNotifier{
topicURL: strings.TrimRight(topicURL, "/"), topicURL: strings.TrimRight(topicURL, "/"),
token: token, token: token,
title: defaultNtfyTitle,
http: &http.Client{Timeout: 10 * time.Second}, http: &http.Client{Timeout: 10 * time.Second},
} }
} }
// WithTitle returns a copy that sends title as the ntfy Title header, so each
// alert source (the capture audit, the symlink tripwire, governance edits)
// is recognisable on the phone (hyperguild#163). A nil notifier (ntfy not
// configured) stays nil.
func (n *NtfyNotifier) WithTitle(title string) *NtfyNotifier {
if n == nil {
return nil
}
c := *n
c.title = title
return &c
}
// Notify posts a message to the ntfy topic. // Notify posts a message to the ntfy topic.
func (n *NtfyNotifier) Notify(ctx context.Context, msg string) error { func (n *NtfyNotifier) Notify(ctx context.Context, msg string) error {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.topicURL, strings.NewReader(msg)) req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.topicURL, strings.NewReader(msg))
if err != nil { if err != nil {
return err return err
} }
req.Header.Set("Title", n.title) req.Header.Set("Title", "brain-capture audit degraded")
req.Header.Set("Priority", "high") req.Header.Set("Priority", "high")
req.Header.Set("Tags", "warning,brain") req.Header.Set("Tags", "warning,brain")
if n.token != "" { if n.token != "" {
@@ -1,29 +0,0 @@
package audit
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// hyperguild#163: every alert went out titled "brain-capture audit
// degraded", including the symlink tripwire's and governance alerts.
func TestNtfyNotifierSendsItsOwnTitle(t *testing.T) {
var got []string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got = append(got, r.Header.Get("Title"))
}))
defer srv.Close()
base := NewNtfyNotifier(srv.URL, "")
require.NoError(t, base.Notify(context.Background(), "m"))
require.NoError(t, base.WithTitle("brain symlink tripwire").Notify(context.Background(), "m"))
require.NoError(t, base.Notify(context.Background(), "m"))
assert.Equal(t, []string{"brain-capture audit degraded", "brain symlink tripwire", "brain-capture audit degraded"}, got,
"WithTitle returns a copy; the base keeps its default title")
var nilN *NtfyNotifier
assert.Nil(t, nilN.WithTitle("x"), "nil (ntfy not configured) stays nil")
}
+22 -1
View File
@@ -109,7 +109,7 @@ func (b *Brain) ReadDirRegular(relDir string) ([]string, error) {
// after following a symlink, and a ".." after a symlinked directory would // after following a symlink, and a ".." after a symlinked directory would
// otherwise pass the check and redirect the read or write. // otherwise pass the check and redirect the read or write.
func cleanRel(rel string) error { func cleanRel(rel string) error {
if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) || strings.Contains(rel, "\\") { if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) {
return fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular) return fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular)
} }
return nil return nil
@@ -243,6 +243,27 @@ func (b *Brain) RealParent(rel string) error {
return b.realDir(path.Dir(rel)) return b.realDir(path.Dir(rel))
} }
// PutFile opens the brain dir as a root and calls PutRegular, for one-off
// writers that hold no Brain.
func PutFile(brainDir, rel string, data []byte) error {
b, err := Open(brainDir)
if err != nil {
return err
}
defer func() { _ = b.Close() }()
return b.PutRegular(rel, data)
}
// AppendFile opens the brain dir as a root and calls AppendRegular.
func AppendFile(brainDir, rel string, data []byte) error {
b, err := Open(brainDir)
if err != nil {
return err
}
defer func() { _ = b.Close() }()
return b.AppendRegular(rel, data)
}
// mkdirReal makes sure every component of rel exists as a real directory, // mkdirReal makes sure every component of rel exists as a real directory,
// creating missing ones through the root; an existing component that is // creating missing ones through the root; an existing component that is
// not a real directory is ErrNotRegular. // not a real directory is ErrNotRegular.
@@ -529,3 +529,20 @@ func TestCreateRegularCreatesOnlyNewFilesThroughRealDirectories(t *testing.T) {
got, _ = os.ReadFile(filepath.Join(dir, "wiki", "new", "facts", "n.md")) got, _ = os.ReadFile(filepath.Join(dir, "wiki", "new", "facts", "n.md"))
assert.Equal(t, "one", string(got)) assert.Equal(t, "one", string(got))
} }
// On Linux a backslash is an ordinary filename byte, not a separator, so a
// name containing one stays a single in-root component (hyperguild#158: the
// watcher retried such a note forever while cleanRel refused it).
func TestReadAndPutRegularTreatABackslashAsAFilenameByte(t *testing.T) {
dir := t.TempDir()
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
require.NoError(t, b.PutRegular(`a\..\b.md`, []byte("x")))
got, err := os.ReadFile(filepath.Join(dir, `a\..\b.md`))
require.NoError(t, err, "the name was not kept as one component")
assert.Equal(t, "x", string(got))
got, err = b.ReadRegular(`a\..\b.md`)
require.NoError(t, err)
assert.Equal(t, "x", string(got))
}
@@ -88,7 +88,6 @@ func TestNoPackageReferencesExtractTextOutsideExtract(t *testing.T) {
// osOpenAllowed names the functions in guarded files that may call os.Open* // osOpenAllowed names the functions in guarded files that may call os.Open*
// on a path that is not a raw/ read. // on a path that is not a raw/ read.
var osOpenAllowed = map[string]bool{ var osOpenAllowed = map[string]bool{
"internal/watcher/watcher.go:appendWatcherLog": true, // ingest log append, not a raw/ read
// Text reads by path; outside this package only TextIn's temp copy reaches // Text reads by path; outside this package only TextIn's temp copy reaches
// it (TestNoPackageReferencesExtractTextOutsideExtract enforces that). // it (TestNoPackageReferencesExtractTextOutsideExtract enforces that).
"internal/extract/extract.go:Text": true, "internal/extract/extract.go:Text": true,
+3 -4
View File
@@ -83,9 +83,8 @@ func (r *Registry) Observe(path, status string, d time.Duration) {
// Middleware wraps next, observing every request's duration + status. // Middleware wraps next, observing every request's duration + status.
// The metric label `path` uses the request's Pattern (Go 1.22+ ServeMux), // The metric label `path` uses the request's Pattern (Go 1.22+ ServeMux),
// or the fixed label "unmatched" if no Pattern is set: falling back to the // falling back to the URL path if no Pattern is set. Pattern keeps
// raw URL path gave any caller one series per unique path (hyperguild#185). // cardinality bounded (one series per route, not one per unique URL).
// Pattern keeps cardinality bounded (one series per route).
func (r *Registry) Middleware(next http.Handler) http.Handler { func (r *Registry) Middleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
rec := &statusRecorder{ResponseWriter: w, code: http.StatusOK} rec := &statusRecorder{ResponseWriter: w, code: http.StatusOK}
@@ -93,7 +92,7 @@ func (r *Registry) Middleware(next http.Handler) http.Handler {
next.ServeHTTP(rec, req) next.ServeHTTP(rec, req)
path := req.Pattern path := req.Pattern
if path == "" { if path == "" {
path = "unmatched" path = req.URL.Path
} }
r.Observe(path, statusClass(rec.code), time.Since(start)) r.Observe(path, statusClass(rec.code), time.Since(start))
}) })
+2 -23
View File
@@ -71,16 +71,11 @@ func TestMiddleware_RecordsTiming(t *testing.T) {
t.Parallel() t.Parallel()
r := New() r := New()
// Routed through a ServeMux, as in production: the mux sets the handler := r.Middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
// request's Pattern, which is the label (an unrouted request is
// "unmatched", hyperguild#185).
mux := http.NewServeMux()
mux.HandleFunc("/query", func(w http.ResponseWriter, _ *http.Request) {
time.Sleep(2 * time.Millisecond) time.Sleep(2 * time.Millisecond)
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
_, _ = io.WriteString(w, "ok") _, _ = io.WriteString(w, "ok")
}) }))
handler := r.Middleware(mux)
srv := httptest.NewServer(handler) srv := httptest.NewServer(handler)
defer srv.Close() defer srv.Close()
@@ -122,19 +117,3 @@ func TestStatusRecorder_DefaultsTo200(t *testing.T) {
t.Errorf("code %d, want 200", rec.Code) t.Errorf("code %d, want 200", rec.Code)
} }
} }
// A request with no matched Pattern is labelled "unmatched", never with its
// raw URL path (unbounded cardinality, hyperguild#185).
func TestMiddleware_LabelsAnUnmatchedRequestWithAFixedLabel(t *testing.T) {
r := New()
h := r.Middleware(http.NotFoundHandler())
h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "/zqrandom-185-unit", nil))
rec := httptest.NewRecorder()
r.Handler().ServeHTTP(rec, httptest.NewRequest("GET", "/metrics", nil))
if strings.Contains(rec.Body.String(), "zqrandom-185-unit") {
t.Errorf("the raw path became a metrics label:\n%s", rec.Body.String())
}
if !strings.Contains(rec.Body.String(), `path="unmatched"`) {
t.Errorf("no unmatched label:\n%s", rec.Body.String())
}
}
-309
View File
@@ -1,309 +0,0 @@
// Package migrate moves pages out of the retired wiki/{sources,concepts,
// entities}/ layout into wiki/<wing>/{sources,facts}/ (hyperguild#205).
//
// It is a two-step tool on purpose. BuildPlan only reads and decides; Apply
// performs exactly the items the plan marked Auto and nothing else. Sources are
// placed by the wing Resolver. Concepts and entities have no wing of their own:
// they inherit the strict-majority wing of the sources that cite them, which is
// deterministic and needs no model. Whatever cannot be placed safely is an
// Exception, left where it is for a human.
package migrate
import (
"context"
"errors"
"fmt"
"io/fs"
"path"
"regexp"
"sort"
"strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
// Status is whether the plan moves a page by itself.
type Status string
const (
Auto Status = "auto"
Exception Status = "exception"
)
// Item is one planned move.
type Item struct {
From string // brain-relative, e.g. wiki/entities/jev.md
To string // brain-relative; empty for an Exception without a proposal
Slug string
Kind string // source | concept | entity
Wing string // decided wing, or the proposal for an Exception
Status Status
Method string // resolver | inherited | junk | blocked
Confidence float64
Reason string
}
// ResolverFor returns the Resolver to use for one page. It takes the page's
// own path so a scorer can exclude the page from its evidence.
type ResolverFor func(excludePath string) *wing.Resolver
var (
linkSlugRE = regexp.MustCompile(`\[\[([^|\]]+)`)
legacy = []struct{ dir, kind string }{
{"wiki/sources", "source"},
{"wiki/concepts", "concept"},
{"wiki/entities", "entity"},
}
)
// BuildPlan reads the legacy layout and decides, without writing anything.
func BuildPlan(ctx context.Context, brainDir string, rf ResolverFor) ([]Item, error) {
bfs, err := brainfs.Open(brainDir)
if err != nil {
return nil, fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
taken, err := slugsInWings(bfs)
if err != nil {
return nil, err
}
var items []Item
decided := map[string]Item{} // source slug -> its decision
for _, l := range legacy {
names, err := bfs.ReadDirRegular(l.dir)
if errors.Is(err, fs.ErrNotExist) {
continue
}
if err != nil {
return nil, fmt.Errorf("read %s: %w", l.dir, err)
}
for _, name := range names {
if !strings.HasSuffix(name, ".md") || name == "_index.md" {
continue
}
rel := l.dir + "/" + name
slug := strings.TrimSuffix(name, ".md")
it := Item{From: rel, Slug: slug, Kind: l.kind}
if strings.HasPrefix(name, "tunnel-candidates-") {
it.Status, it.Method, it.Reason = Exception, "junk", "auto-tunnel review log ingested as a source; not a source"
items = append(items, it)
continue
}
content, err := bfs.ReadRegular(rel)
if err != nil {
it.Status, it.Method, it.Reason = Exception, "blocked", fmt.Sprintf("unreadable: %v", err)
items = append(items, it)
continue
}
if l.kind == "source" {
d := rf(rel).Resolve(ctx, wing.Note{Text: string(content), FrontmatterWing: frontmatterValue(string(content), "wing")})
it.Method, it.Wing, it.Confidence, it.Reason = "resolver", d.Wing, d.Confidence, d.Reason
if d.Source == wing.Exception {
it.Status = Exception
} else {
it.Status, it.To = Auto, "wiki/"+d.Wing+"/sources/"+name
}
decided[slug] = it
}
items = append(items, it)
}
}
// Second pass: concepts and entities inherit from their sources.
for i, it := range items {
if it.Kind == "source" || it.Method == "junk" || it.Method == "blocked" {
continue
}
content, err := bfs.ReadRegular(it.From)
if err != nil {
continue
}
it.Method = "inherited"
w, reason := inheritedWing(string(content), decided)
if w == "" {
it.Status, it.Reason = Exception, reason
} else {
it.Status, it.Wing, it.To = Auto, w, "wiki/"+w+"/facts/"+it.Slug+".md"
it.Reason = fmt.Sprintf("strict majority of its sources are in %q", w)
}
items[i] = it
}
blockCollisions(bfs, items, taken)
sort.SliceStable(items, func(i, j int) bool { return items[i].From < items[j].From })
return items, nil
}
// inheritedWing returns the wing held by a strict majority of the legacy
// sources cited in the page's "## Sources" section, or "" with the reason.
func inheritedWing(content string, decided map[string]Item) (string, string) {
section := sourcesSection(content)
if section == "" {
return "", "page has no Sources section to inherit a wing from"
}
total := 0
votes := map[string]int{}
for _, m := range linkSlugRE.FindAllStringSubmatch(section, -1) {
d, ok := decided[strings.TrimSpace(m[1])]
if !ok {
continue // not a legacy source (already migrated, or not a source)
}
total++
if d.Status == Auto {
votes[d.Wing]++
}
}
if total == 0 {
return "", "none of its cited sources are in the legacy layout"
}
for w, n := range votes {
if n*2 > total {
return w, ""
}
}
return "", fmt.Sprintf("its %d cited sources disagree on a wing or were not placed", total)
}
// blockCollisions turns an Auto item into an Exception when its destination
// exists, its slug is already used in some wing (graph identity is the slug
// alone), or two planned items would land on the same path.
func blockCollisions(bfs *brainfs.Brain, items []Item, taken map[string]string) {
planned := map[string]int{}
for _, it := range items {
if it.Status == Auto {
planned[it.To]++
}
}
for i, it := range items {
if it.Status != Auto {
continue
}
switch {
case planned[it.To] > 1:
items[i].Status, items[i].Reason = Exception, fmt.Sprintf("another legacy page would land on %s", it.To)
case exists(bfs, it.To):
items[i].Status, items[i].Reason = Exception, fmt.Sprintf("%s already exists", it.To)
default:
if other, ok := taken[it.Slug]; ok {
items[i].Status, items[i].Reason = Exception, fmt.Sprintf("slug %q is already used by %s", it.Slug, other)
}
}
}
}
func exists(bfs *brainfs.Brain, rel string) bool {
_, err := bfs.Lstat(rel)
return err == nil
}
// slugsInWings maps every slug under wiki/<wing>/<hall>/ to its path.
func slugsInWings(bfs *brainfs.Brain) (map[string]string, error) {
out := map[string]string{}
wings, err := bfs.ReadDirDirs("wiki")
if errors.Is(err, fs.ErrNotExist) {
return out, nil
}
if err != nil {
return nil, fmt.Errorf("list wiki: %w", err)
}
for _, w := range wings {
if w == "sources" || w == "concepts" || w == "entities" {
continue
}
halls, err := bfs.ReadDirDirs("wiki/" + w)
if err != nil {
continue
}
for _, h := range halls {
names, err := bfs.ReadDirRegular("wiki/" + w + "/" + h)
if err != nil {
continue
}
for _, n := range names {
if strings.HasSuffix(n, ".md") && n != "_index.md" {
out[strings.TrimSuffix(n, ".md")] = "wiki/" + w + "/" + h + "/" + n
}
}
}
}
return out, nil
}
// Apply performs the Auto items, one at a time, and stops at the first
// failure. Each move writes the destination with O_EXCL, then removes the
// source, so a failure never loses a page. Exceptions are not touched.
func Apply(brainDir string, items []Item) (int, error) {
bfs, err := brainfs.Open(brainDir)
if err != nil {
return 0, fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
n := 0
touched := map[string]bool{}
for _, it := range items {
if it.Status != Auto {
continue
}
content, err := bfs.ReadRegular(it.From)
if err != nil {
return n, fmt.Errorf("read %s: %w", it.From, err)
}
hall := path.Base(path.Dir(it.To))
if err := bfs.CreateRegular(it.To, []byte(stamp(string(content), it.Wing, hall, it.Kind))); err != nil {
return n, fmt.Errorf("write %s: %w", it.To, err)
}
if err := bfs.Remove(it.From); err != nil {
return n, fmt.Errorf("moved to %s but removing %s failed: %w", it.To, it.From, err)
}
touched[it.Wing] = true
n++
}
for w := range touched {
_ = brain.BuildWingIndex(brainDir, w) // best-effort upkeep; the pages are already moved
}
return n, nil
}
// stamp inserts wing/hall (and kind for facts) after the opening fence, leaving
// every other byte of the page alone. A page without frontmatter gets a block.
func stamp(content, w, hall, kind string) string {
extra := "wing: " + w + "\nhall: " + hall + "\n"
if hall == "facts" {
extra += "kind: " + kind + "\n"
}
if strings.HasPrefix(content, "---\n") {
return "---\n" + extra + content[4:]
}
return "---\n" + extra + "---\n" + content
}
// sourcesSection returns the text under a "## Sources" heading up to the next
// "## " heading.
func sourcesSection(content string) string {
_, rest, ok := strings.Cut(content, "\n## Sources")
if !ok {
return ""
}
if i := strings.Index(rest, "\n## "); i >= 0 {
rest = rest[:i]
}
return rest
}
func frontmatterValue(content, key string) string {
if !strings.HasPrefix(content, "---\n") {
return ""
}
block, _, _ := strings.Cut(content[4:], "\n---")
for _, line := range strings.Split(block, "\n") {
if k, v, ok := strings.Cut(line, ":"); ok && strings.TrimSpace(k) == key {
return strings.Trim(strings.TrimSpace(v), `"'`)
}
}
return ""
}
-243
View File
@@ -1,243 +0,0 @@
package migrate_test
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/migrate"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
// keywordClassifier picks a wing by a marker word in the text, and the scorer
// agrees by the same marker, so a page "belongs" to the wing whose marker it holds.
type keywordClassifier struct{}
func (keywordClassifier) Classify(_ context.Context, text string, _ []string) (wing.Suggestion, error) {
for _, w := range []string{"alpha", "beta"} {
if strings.Contains(text, strings.ToUpper(w)) {
return wing.Suggestion{Wing: w, Confidence: 0.95, Reason: "marker"}, nil
}
}
return wing.Suggestion{}, nil
}
type keywordScorer struct{}
func (keywordScorer) Score(text string) map[string]float64 {
out := map[string]float64{}
for _, w := range []string{"alpha", "beta"} {
if strings.Contains(text, strings.ToUpper(w)) {
out[w] = 1
}
}
return out
}
func resolverFor(string) *wing.Resolver {
return &wing.Resolver{
Known: []string{"alpha", "beta", "client"},
Eligible: func(w string) bool { return w != "client" },
Classifier: keywordClassifier{},
Scorer: keywordScorer{},
Threshold: 0.8,
}
}
func page(t *testing.T, dir, rel, body string) {
t.Helper()
full := filepath.Join(dir, rel)
require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755))
require.NoError(t, os.WriteFile(full, []byte(body), 0o644))
}
func src(title, marker string) string {
return "---\ntitle: " + title + "\ntype: video\n---\n\n## Summary\n\n" + marker + " content.\n"
}
func entity(title string, sources ...string) string {
var b strings.Builder
b.WriteString("---\ntitle: " + title + "\ntype: tool\n---\n\n## Definition\n\nA thing.\n\n## Sources\n\n")
for _, s := range sources {
b.WriteString("- [[" + s + "|" + s + "]]\n")
}
return b.String()
}
func byFrom(items []migrate.Item) map[string]migrate.Item {
m := map[string]migrate.Item{}
for _, it := range items {
m[it.From] = it
}
return m
}
func TestPlan_AResolvedSourceGoesToItsWingsSourcesHall(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/talk-a.md", src("Talk A", "ALPHA"))
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
it := byFrom(items)["wiki/sources/talk-a.md"]
assert.Equal(t, migrate.Auto, it.Status)
assert.Equal(t, "wiki/alpha/sources/talk-a.md", it.To)
}
func TestPlan_AutoTunnelLogsAreJunkNotSources(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/tunnel-candidates-2026-06-14.md", src("t", "ALPHA"))
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
it := byFrom(items)["wiki/sources/tunnel-candidates-2026-06-14.md"]
assert.Equal(t, migrate.Exception, it.Status, "never auto-moved, whatever the classifier says")
assert.Equal(t, "junk", it.Method)
}
func TestPlan_AnEntityFollowsTheMajorityWingOfItsSources(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/a1.md", src("A1", "ALPHA"))
page(t, dir, "wiki/sources/a2.md", src("A2", "ALPHA"))
page(t, dir, "wiki/sources/b1.md", src("B1", "BETA"))
page(t, dir, "wiki/entities/jev.md", entity("Jev", "a1", "a2", "b1"))
page(t, dir, "wiki/concepts/factory.md", "---\ntitle: Factory\n---\n\n## Definition\n\nx\n\n## Sources\n\n- [[a1|A1]]\n")
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
m := byFrom(items)
assert.Equal(t, "wiki/alpha/facts/jev.md", m["wiki/entities/jev.md"].To)
assert.Equal(t, "entity", m["wiki/entities/jev.md"].Kind)
assert.Equal(t, "wiki/alpha/facts/factory.md", m["wiki/concepts/factory.md"].To)
assert.Equal(t, "concept", m["wiki/concepts/factory.md"].Kind)
}
func TestPlan_ATieBetweenWingsIsAnException(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/a1.md", src("A1", "ALPHA"))
page(t, dir, "wiki/sources/b1.md", src("B1", "BETA"))
page(t, dir, "wiki/entities/jev.md", entity("Jev", "a1", "b1"))
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
it := byFrom(items)["wiki/entities/jev.md"]
assert.Equal(t, migrate.Exception, it.Status)
assert.Contains(t, it.Reason, "disagree")
}
func TestPlan_AnEntityWhoseSourcesWereNotPlacedIsAnException(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/mystery.md", src("Mystery", "nothing recognisable"))
page(t, dir, "wiki/entities/orphan.md", entity("Orphan", "mystery"))
page(t, dir, "wiki/entities/nosources.md", "---\ntitle: No Sources\n---\n\n## Definition\n\nx\n")
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
m := byFrom(items)
assert.Equal(t, migrate.Exception, m["wiki/entities/orphan.md"].Status)
assert.Equal(t, migrate.Exception, m["wiki/entities/nosources.md"].Status)
}
func TestPlan_ADestinationThatAlreadyExistsIsAnException(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/talk-a.md", src("Talk A", "ALPHA"))
page(t, dir, "wiki/alpha/sources/talk-a.md", "someone else's page")
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
it := byFrom(items)["wiki/sources/talk-a.md"]
assert.Equal(t, migrate.Exception, it.Status)
assert.Contains(t, it.Reason, "already exists", "the occupied destination is named as the reason, not a generic slug clash")
}
func TestPlan_ASlugAlreadyUsedInAnotherWingIsAnException(t *testing.T) {
// Graph identity is the slug alone; two pages with one slug would merge.
dir := t.TempDir()
page(t, dir, "wiki/sources/talk-a.md", src("Talk A", "ALPHA"))
page(t, dir, "wiki/beta/facts/talk-a.md", "a beta page with the same slug")
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
it := byFrom(items)["wiki/sources/talk-a.md"]
assert.Equal(t, migrate.Exception, it.Status)
assert.Contains(t, it.Reason, "slug")
}
func TestApply_MovesOnlyAutoItemsAndStampsFrontmatterWithoutTouchingTheBody(t *testing.T) {
dir := t.TempDir()
body := src("Talk A", "ALPHA") + "\nSee [[jev|Jev]] and [[unknown-slug]].\n"
page(t, dir, "wiki/sources/talk-a.md", body)
page(t, dir, "wiki/sources/tunnel-candidates-2026-06-14.md", src("t", "ALPHA"))
page(t, dir, "wiki/entities/jev.md", entity("Jev", "talk-a"))
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
n, err := migrate.Apply(dir, items)
require.NoError(t, err)
assert.Equal(t, 2, n)
moved, err := os.ReadFile(filepath.Join(dir, "wiki/alpha/sources/talk-a.md"))
require.NoError(t, err)
assert.Contains(t, string(moved), "wing: alpha")
assert.Contains(t, string(moved), "hall: sources")
assert.Contains(t, string(moved), "See [[jev|Jev]] and [[unknown-slug]].", "links and body are byte-preserved")
ent, err := os.ReadFile(filepath.Join(dir, "wiki/alpha/facts/jev.md"))
require.NoError(t, err)
assert.Contains(t, string(ent), "kind: entity")
_, err = os.Stat(filepath.Join(dir, "wiki/sources/talk-a.md"))
assert.ErrorIs(t, err, os.ErrNotExist, "the original is removed after the copy is written")
_, err = os.Stat(filepath.Join(dir, "wiki/sources/tunnel-candidates-2026-06-14.md"))
assert.NoError(t, err, "an exception is left exactly where it is")
}
func TestApply_NeverOverwritesAndKeepsTheSourceWhenTheDestinationAppearsLate(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/talk-a.md", src("Talk A", "ALPHA"))
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
page(t, dir, "wiki/alpha/sources/talk-a.md", "appeared after the plan")
n, err := migrate.Apply(dir, items)
require.Error(t, err)
assert.Zero(t, n)
got, _ := os.ReadFile(filepath.Join(dir, "wiki/alpha/sources/talk-a.md"))
assert.Equal(t, "appeared after the plan", string(got))
_, statErr := os.Stat(filepath.Join(dir, "wiki/sources/talk-a.md"))
assert.NoError(t, statErr, "the source survives a failed write")
}
func TestPlan_IsEmptyOnceEverythingIsMigrated(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/talk-a.md", src("Talk A", "ALPHA"))
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
_, err = migrate.Apply(dir, items)
require.NoError(t, err)
again, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
assert.Empty(t, again)
}
// An exception that blocked on a collision still carries a destination; Apply
// must treat Status, not the presence of a path, as the decision.
func TestApply_LeavesACollisionExceptionAloneEvenThoughItHasADestination(t *testing.T) {
dir := t.TempDir()
page(t, dir, "wiki/sources/talk-a.md", src("Talk A", "ALPHA"))
page(t, dir, "wiki/alpha/sources/talk-a.md", "occupied")
items, err := migrate.BuildPlan(context.Background(), dir, resolverFor)
require.NoError(t, err)
require.Equal(t, migrate.Exception, byFrom(items)["wiki/sources/talk-a.md"].Status)
n, err := migrate.Apply(dir, items)
require.NoError(t, err)
assert.Zero(t, n)
_, statErr := os.Stat(filepath.Join(dir, "wiki/sources/talk-a.md"))
assert.NoError(t, statErr)
}
+1 -4
View File
@@ -7,8 +7,6 @@ import (
"fmt" "fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"io/fs" "io/fs"
"os"
"path/filepath"
"strings" "strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki" "github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
@@ -90,8 +88,7 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
} }
for relPath, page := range pending { for relPath, page := range pending {
dest := filepath.Join(brainDir, filepath.FromSlash(relPath)) if err := bfs.PutRegular(relPath, []byte(page.Content)); err != nil {
if err := os.WriteFile(dest, []byte(page.Content), 0o644); err != nil {
return 0, fmt.Errorf("write %s: %w", relPath, err) return 0, fmt.Errorf("write %s: %w", relPath, err)
} }
} }
+2 -32
View File
@@ -14,13 +14,6 @@ import (
// (derived from the filename, not the LLM title). Pages whose title resolves to an // (derived from the filename, not the LLM title). Pages whose title resolves to an
// empty slug are skipped and returned as warnings instead. // empty slug are skipped and returned as warnings instead.
func BuildPages(rawPages []RawPage, sourceSlug, date string) ([]wiki.Page, []string) { func BuildPages(rawPages []RawPage, sourceSlug, date string) ([]wiki.Page, []string) {
return BuildPagesIn("", rawPages, sourceSlug, date)
}
// BuildPagesIn is BuildPages for a wing: sources go to wiki/<wing>/sources/,
// concepts and entities to wiki/<wing>/facts/ with `kind:` recording which.
// An empty wing is the retired wiki/{sources,concepts,entities}/ layout.
func BuildPagesIn(wing string, rawPages []RawPage, sourceSlug, date string) ([]wiki.Page, []string) {
out := make([]wiki.Page, 0, len(rawPages)) out := make([]wiki.Page, 0, len(rawPages))
var warnings []string var warnings []string
for _, rp := range rawPages { for _, rp := range rawPages {
@@ -35,7 +28,7 @@ func BuildPagesIn(wing string, rawPages []RawPage, sourceSlug, date string) ([]w
warnings = append(warnings, fmt.Sprintf("skipped page with empty title (type: %s)", rp.Type)) warnings = append(warnings, fmt.Sprintf("skipped page with empty title (type: %s)", rp.Type))
continue continue
} }
out = append(out, buildPage(rp, sourceSlug, date, wing)) out = append(out, buildPage(rp, sourceSlug, date))
} }
return out, warnings return out, warnings
} }
@@ -49,7 +42,7 @@ func computeSlug(rp RawPage, sourceSlug string) string {
return wiki.Slug(rp.Title) return wiki.Slug(rp.Title)
} }
func buildPage(rp RawPage, sourceSlug, date, wing string) wiki.Page { func buildPage(rp RawPage, sourceSlug, date string) wiki.Page {
var slug, dir string var slug, dir string
switch rp.Type { switch rp.Type {
case "source": case "source":
@@ -66,21 +59,8 @@ func buildPage(rp RawPage, sourceSlug, date, wing string) wiki.Page {
dir = "wiki/" + rp.Type dir = "wiki/" + rp.Type
} }
hall := ""
if wing != "" {
switch rp.Type {
case "source":
hall = "sources"
default:
hall = "facts"
}
dir = "wiki/" + wing + "/" + hall
}
path := dir + "/" + slug + ".md" path := dir + "/" + slug + ".md"
fm := buildFrontmatter(rp, date) fm := buildFrontmatter(rp, date)
if wing != "" {
fm = addWingFrontmatter(fm, wing, hall, rp.Type)
}
return wiki.Page{ return wiki.Page{
Path: path, Path: path,
@@ -142,16 +122,6 @@ func buildFrontmatter(rp RawPage, date string) string {
return sb.String() return sb.String()
} }
// addWingFrontmatter inserts wing/hall (and kind for facts pages) right after
// the opening fence of an already-built frontmatter block.
func addWingFrontmatter(fm, wing, hall, rpType string) string {
extra := "wing: " + wing + "\nhall: " + hall + "\n"
if hall == "facts" {
extra += "kind: " + rpType + "\n"
}
return strings.Replace(fm, "---\n", "---\n"+extra, 1)
}
func yamlScalar(s string) string { func yamlScalar(s string) string {
// A raw newline or CR would end the scalar for every line-oriented // A raw newline or CR would end the scalar for every line-oriented
// frontmatter reader and start a new key (PR #147 review: a title of // frontmatter reader and start a new key (PR #147 review: a title of
+26 -42
View File
@@ -7,11 +7,8 @@ import (
"encoding/hex" "encoding/hex"
"errors" "errors"
"fmt" "fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"io/fs" "io/fs"
"os"
"path/filepath"
"strings" "strings"
"time" "time"
@@ -27,11 +24,6 @@ type Config struct {
Complete CompleteFunc Complete CompleteFunc
ChunkSize int // 0 = no chunking ChunkSize int // 0 = no chunking
Schema string // overrides brain/schema.md when set (useful in tests) Schema string // overrides brain/schema.md when set (useful in tests)
// Wing, when set, writes into wiki/<wing>/{sources,facts}/ and scopes the
// inventory (title resolution, back-references) to that wing alone. Empty
// keeps the retired wiki/{sources,concepts,entities}/ layout.
Wing string
} }
// Result is the outcome of a pipeline run. // Result is the outcome of a pipeline run.
@@ -48,16 +40,7 @@ type Result struct {
// Run ingests content and writes structured wiki pages to brainDir/wiki/. // Run ingests content and writes structured wiki pages to brainDir/wiki/.
// In dry-run mode, pages are returned but not written to disk. // In dry-run mode, pages are returned but not written to disk.
func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryRun bool) (Result, error) { func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryRun bool) (Result, error) {
if cfg.Wing != "" && brain.Sanitise(cfg.Wing) != cfg.Wing { inventory, err := wiki.LoadInventory(brainDir)
return Result{}, fmt.Errorf("invalid wing %q: must be a plain lower-case wing name", cfg.Wing)
}
var inventory map[wiki.PageType][]wiki.Entry
var err error
if cfg.Wing != "" {
inventory, err = wiki.LoadWingInventory(brainDir, cfg.Wing)
} else {
inventory, err = wiki.LoadInventory(brainDir)
}
if err != nil { if err != nil {
return Result{}, fmt.Errorf("load inventory: %w", err) return Result{}, fmt.Errorf("load inventory: %w", err)
} }
@@ -75,7 +58,7 @@ func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryR
// The merge is already idempotent (#103), so this is cost and signal, not // The merge is already idempotent (#103), so this is cost and signal, not
// correctness — which is why it is safe for it to be an early return. // correctness — which is why it is safe for it to be an early return.
bodyHash := hashBody(content) bodyHash := hashBody(content)
if !dryRun && bodyHash != "" && bodyHash == storedSourceHash(brainDir, cfg.Wing, sourceSlug) { if !dryRun && bodyHash != "" && bodyHash == storedSourceHash(brainDir, sourceSlug) {
return Result{Skipped: true, Warnings: []string{ return Result{Skipped: true, Warnings: []string{
fmt.Sprintf("source %q unchanged since last ingest (hash %s) — extraction skipped", source, bodyHash[:12]), fmt.Sprintf("source %q unchanged since last ingest (hash %s) — extraction skipped", source, bodyHash[:12]),
}}, nil }}, nil
@@ -86,7 +69,7 @@ func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryR
// only exists once the model has produced a title. So page-aware // only exists once the model has produced a title. So page-aware
// extraction (#106) is scoped to the source page: it's the one page every // extraction (#106) is scoped to the source page: it's the one page every
// chunk's prompt can be given in advance. // chunk's prompt can be given in advance.
existingPage := readSourcePage(brainDir, cfg.Wing, sourceSlug) existingPage := readSourcePage(brainDir, sourceSlug)
chunks := Chunk(content, cfg.ChunkSize) chunks := Chunk(content, cfg.ChunkSize)
@@ -108,7 +91,7 @@ func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryR
meta.BodyHash = bodyHash meta.BodyHash = bodyHash
applySourceMeta(allRaw, meta) applySourceMeta(allRaw, meta)
return buildAndWrite(ctx, cfg.Wing, allRaw, sourceSlug, date, brainDir, source, inventory, allWarnings, dryRun) return buildAndWrite(ctx, allRaw, sourceSlug, date, brainDir, source, inventory, allWarnings, dryRun)
} }
// RunRaw runs the pipeline on pre-parsed RawPages, skipping the LLM extraction // RunRaw runs the pipeline on pre-parsed RawPages, skipping the LLM extraction
@@ -127,12 +110,12 @@ func RunRaw(ctx context.Context, brainDir, source string, rawPages []RawPage, dr
sourceSlug := wiki.Slug(source) sourceSlug := wiki.Slug(source)
date := time.Now().UTC().Format("2006-01-02") date := time.Now().UTC().Format("2006-01-02")
return buildAndWrite(ctx, "", rawPages, sourceSlug, date, brainDir, source, inventory, nil, dryRun) return buildAndWrite(ctx, rawPages, sourceSlug, date, brainDir, source, inventory, nil, dryRun)
} }
// buildAndWrite runs BuildPages through write for both Run and RunRaw. // buildAndWrite runs BuildPages through write for both Run and RunRaw.
func buildAndWrite(ctx context.Context, wing string, rawPages []RawPage, sourceSlug, date, brainDir, source string, inventory map[wiki.PageType][]wiki.Entry, warnings []string, dryRun bool) (Result, error) { func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, brainDir, source string, inventory map[wiki.PageType][]wiki.Entry, warnings []string, dryRun bool) (Result, error) {
pages, buildWarnings := BuildPagesIn(wing, rawPages, sourceSlug, date) pages, buildWarnings := BuildPages(rawPages, sourceSlug, date)
warnings = append(warnings, buildWarnings...) warnings = append(warnings, buildWarnings...)
resolved := Resolve(pages, inventory) resolved := Resolve(pages, inventory)
canonicalized, linkWarnings := CanonicalizeLinks(resolved, inventory) canonicalized, linkWarnings := CanonicalizeLinks(resolved, inventory)
@@ -143,10 +126,6 @@ func buildAndWrite(ctx context.Context, wing string, rawPages []RawPage, sourceS
var written []string var written []string
for _, page := range merged { for _, page := range merged {
if !dryRun { if !dryRun {
dest := filepath.Join(brainDir, filepath.FromSlash(page.Path))
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err)
}
content, conflicts, err := reviseOrCreate(brainDir, page) content, conflicts, err := reviseOrCreate(brainDir, page)
if err != nil { if err != nil {
return Result{}, err return Result{}, err
@@ -165,7 +144,9 @@ func buildAndWrite(ctx context.Context, wing string, rawPages []RawPage, sourceS
if err != nil { if err != nil {
return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err) return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err)
} }
if err := os.WriteFile(dest, []byte(stamped), 0o644); err != nil { // Written through the brain root, never through a symlinked
// directory or page (hyperguild#158).
if err := writeBrainFile(brainDir, page.Path, []byte(stamped)); err != nil {
return Result{}, fmt.Errorf("write %s: %w", page.Path, err) return Result{}, fmt.Errorf("write %s: %w", page.Path, err)
} }
} }
@@ -173,11 +154,7 @@ func buildAndWrite(ctx context.Context, wing string, rawPages []RawPage, sourceS
} }
if !dryRun { if !dryRun {
if wing != "" { if err := wiki.RebuildIndex(brainDir, date); err != nil {
if err := brain.BuildWingIndex(brainDir, wing); err != nil {
warnings = append(warnings, fmt.Sprintf("rebuild wing index: %v", err))
}
} else if err := wiki.RebuildIndex(brainDir, date); err != nil {
warnings = append(warnings, fmt.Sprintf("rebuild index: %v", err)) warnings = append(warnings, fmt.Sprintf("rebuild index: %v", err))
} }
if err := wiki.AppendLog(brainDir, source, written, warnings, date); err != nil { if err := wiki.AppendLog(brainDir, source, written, warnings, date); err != nil {
@@ -212,18 +189,25 @@ func hashBody(content string) string {
// wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet. // wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet.
// It reads through the brain root, regular files only (hyperguild#157): the // It reads through the brain root, regular files only (hyperguild#157): the
// content goes into the LLM prompt. // content goes into the LLM prompt.
func readSourcePage(brainDir, wing, sourceSlug string) string { func readSourcePage(brainDir, sourceSlug string) string {
rel := "wiki/sources/" + sourceSlug + ".md" raw, err := readBrainFile(brainDir, "wiki/sources/"+sourceSlug+".md")
if wing != "" {
rel = "wiki/" + wing + "/sources/" + sourceSlug + ".md"
}
raw, err := readBrainFile(brainDir, rel)
if err != nil { if err != nil {
return "" return ""
} }
return string(raw) return string(raw)
} }
// writeBrainFile creates or replaces one brain-relative file through the
// brain root (hyperguild#158).
func writeBrainFile(brainDir, rel string, data []byte) error {
bfs, err := brainfs.Open(brainDir)
if err != nil {
return err
}
defer func() { _ = bfs.Close() }()
return bfs.PutRegular(rel, data)
}
// readBrainFile reads one brain-relative file through the brain root, // readBrainFile reads one brain-relative file through the brain root,
// regular files only (hyperguild#157). // regular files only (hyperguild#157).
func readBrainFile(brainDir, rel string) ([]byte, error) { func readBrainFile(brainDir, rel string) ([]byte, error) {
@@ -241,8 +225,8 @@ func readBrainFile(brainDir, rel string) ([]byte, error) {
// Absence always means "extract". That is the fail-safe direction: a missed // Absence always means "extract". That is the fail-safe direction: a missed
// skip costs one extraction, while a wrong skip loses an update silently — and // skip costs one extraction, while a wrong skip loses an update silently — and
// silence is the failure mode this repo keeps paying for. // silence is the failure mode this repo keeps paying for.
func storedSourceHash(brainDir, wing, sourceSlug string) string { func storedSourceHash(brainDir, sourceSlug string) string {
for _, line := range strings.Split(readSourcePage(brainDir, wing, sourceSlug), "\n") { for _, line := range strings.Split(readSourcePage(brainDir, sourceSlug), "\n") {
if key, value, found := strings.Cut(line, ":"); found && strings.TrimSpace(key) == "source_hash" { if key, value, found := strings.Cut(line, ":"); found && strings.TrimSpace(key) == "source_hash" {
return strings.TrimSpace(value) return strings.TrimSpace(value)
} }
+5 -30
View File
@@ -23,7 +23,7 @@ func injectSourceRefs(pages []wiki.Page, inventory map[wiki.PageType][]wiki.Entr
var sourceContent string var sourceContent string
for _, p := range pages { for _, p := range pages {
if isSourcePath(p.Path) && if strings.HasPrefix(p.Path, "wiki/sources/") &&
strings.TrimSuffix(filepath.Base(p.Path), ".md") == sourceSlug { strings.TrimSuffix(filepath.Base(p.Path), ".md") == sourceSlug {
sourceContent = p.Content sourceContent = p.Content
break break
@@ -35,7 +35,7 @@ func injectSourceRefs(pages []wiki.Page, inventory map[wiki.PageType][]wiki.Entr
bySlug := make(map[string]int, len(pages)) bySlug := make(map[string]int, len(pages))
for i, p := range pages { for i, p := range pages {
if !isSourcePath(p.Path) { if !strings.HasPrefix(p.Path, "wiki/sources/") {
bySlug[strings.TrimSuffix(filepath.Base(p.Path), ".md")] = i bySlug[strings.TrimSuffix(filepath.Base(p.Path), ".md")] = i
} }
} }
@@ -52,13 +52,12 @@ func injectSourceRefs(pages []wiki.Page, inventory map[wiki.PageType][]wiki.Entr
if !ok { if !ok {
continue continue
} }
path := entryPath(slug, pt, inventory) b, err := readBrainFile(brainDir, "wiki/"+string(pt)+"/"+slug+".md")
b, err := readBrainFile(brainDir, path)
if err != nil { if err != nil {
continue continue
} }
page := wiki.Page{ page := wiki.Page{
Path: path, Path: "wiki/" + string(pt) + "/" + slug + ".md",
Content: string(b), Content: string(b),
} }
pages = append(pages, addSourceRef(page, sourceRef)) pages = append(pages, addSourceRef(page, sourceRef))
@@ -89,7 +88,7 @@ func extractWikilinks(content string) map[string]bool {
// findSourcePage returns the slug and title of the first wiki/sources/ page in pages. // findSourcePage returns the slug and title of the first wiki/sources/ page in pages.
func findSourcePage(pages []wiki.Page) (slug, title string, found bool) { func findSourcePage(pages []wiki.Page) (slug, title string, found bool) {
for _, p := range pages { for _, p := range pages {
if isSourcePath(p.Path) { if strings.HasPrefix(p.Path, "wiki/sources/") {
slug = strings.TrimSuffix(filepath.Base(p.Path), ".md") slug = strings.TrimSuffix(filepath.Base(p.Path), ".md")
title = extractTitle(p.Content) title = extractTitle(p.Content)
if title == "" { if title == "" {
@@ -112,27 +111,3 @@ func findInInventory(slug string, inventory map[wiki.PageType][]wiki.Entry) (wik
} }
return "", false return "", false
} }
// isSourcePath reports whether path is a source page in either layout:
// wiki/sources/<slug>.md or wiki/<wing>/sources/<slug>.md.
func isSourcePath(path string) bool {
parts := strings.Split(filepath.ToSlash(path), "/")
switch {
case len(parts) == 3 && parts[0] == "wiki" && parts[1] == "sources":
return true
case len(parts) == 4 && parts[0] == "wiki" && parts[2] == "sources":
return true
}
return false
}
// entryPath is the brain-relative path of an inventory page: the recorded Path
// for a wing-layout entry, else the retired wiki/<type>/<slug>.md.
func entryPath(slug string, pt wiki.PageType, inventory map[wiki.PageType][]wiki.Entry) string {
for _, e := range inventory[pt] {
if e.Slug == slug && e.Path != "" {
return e.Path
}
}
return "wiki/" + string(pt) + "/" + slug + ".md"
}
+1 -17
View File
@@ -32,7 +32,7 @@ func Resolve(proposed []wiki.Page, inventory map[wiki.PageType][]wiki.Entry) []w
out := make([]wiki.Page, 0, len(proposed)) out := make([]wiki.Page, 0, len(proposed))
for _, page := range proposed { for _, page := range proposed {
pt := pageTypeOf(page) pt := pageTypeFromPath(page.Path)
title := extractTitle(page.Content) title := extractTitle(page.Content)
k := key{pt: pt, normalized: normalizeTitle(title)} k := key{pt: pt, normalized: normalizeTitle(title)}
if canonicalSlug, ok := lookup[k]; ok { if canonicalSlug, ok := lookup[k]; ok {
@@ -55,22 +55,6 @@ func normalizeTitle(s string) string {
return strings.Join(strings.Fields(s), " ") return strings.Join(strings.Fields(s), " ")
} }
// pageTypeOf is the page's type: from its directory in the retired layout, or
// from `kind:` for a wing-layout facts page (concepts and entities share facts/).
func pageTypeOf(p wiki.Page) wiki.PageType {
parts := strings.Split(filepath.ToSlash(p.Path), "/")
if len(parts) >= 4 && parts[0] == "wiki" {
switch parts[2] {
case "sources":
return wiki.PageTypeSource
case "facts":
pt, _ := wiki.KindFromContent(p.Content)
return pt
}
}
return pageTypeFromPath(p.Path)
}
// pageTypeFromPath extracts the wiki.PageType from a path like "wiki/entities/foo.md". // pageTypeFromPath extracts the wiki.PageType from a path like "wiki/entities/foo.md".
func pageTypeFromPath(path string) wiki.PageType { func pageTypeFromPath(path string) wiki.PageType {
parts := strings.Split(filepath.ToSlash(path), "/") parts := strings.Split(filepath.ToSlash(path), "/")
+55 -5
View File
@@ -35,17 +35,17 @@ func TestReadSourcePageAndSchemaRefuseSymlinksAndSurviveSwapRaces(t *testing.T)
inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md")) inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok.md"), []byte("zqinside source"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok.md"), []byte("zqinside source"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(brain, "schema.md"), []byte("zqinside schema"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(brain, "schema.md"), []byte("zqinside schema"), 0o644))
assert.Equal(t, "zqinside source", readSourcePage(brain, "", "ok"), "positive control") assert.Equal(t, "zqinside source", readSourcePage(brain, "ok"), "positive control")
assert.Equal(t, "zqinside schema", loadSchema(brain), "positive control") assert.Equal(t, "zqinside schema", loadSchema(brain), "positive control")
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(brain, "wiki", "sources", "out.md"))) require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(brain, "wiki", "sources", "out.md")))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "sources", "in.md"))) require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "sources", "in.md")))
assert.NotContains(t, readSourcePage(brain, "", "out"), brainfstest.Sentinel) assert.NotContains(t, readSourcePage(brain, "out"), brainfstest.Sentinel)
assert.NotContains(t, readSourcePage(brain, "", "in"), brainfstest.Sentinel) assert.NotContains(t, readSourcePage(brain, "in"), brainfstest.Sentinel)
assert.NotContains(t, storedSourceHash(brain, "", "in"), brainfstest.Sentinel) assert.NotContains(t, storedSourceHash(brain, "in"), brainfstest.Sentinel)
src := filepath.Join(brain, "wiki", "sources", "r.md") src := filepath.Join(brain, "wiki", "sources", "r.md")
read := func() string { return readSourcePage(brain, "", "r") } read := func() string { return readSourcePage(brain, "r") }
brainfstest.AssertNoLeak(t, src, "zqinside", 1500*time.Millisecond, read) brainfstest.AssertNoLeak(t, src, "zqinside", 1500*time.Millisecond, read)
brainfstest.AssertNoLeakFrom(t, src, "zqinside", inBrain, 1500*time.Millisecond, read) brainfstest.AssertNoLeakFrom(t, src, "zqinside", inBrain, 1500*time.Millisecond, read)
brainfstest.AssertNoHang(t, src, "zqinside", 1500*time.Millisecond, 6*time.Second, func() { _ = read() }) brainfstest.AssertNoHang(t, src, "zqinside", 1500*time.Millisecond, 6*time.Second, func() { _ = read() })
@@ -165,3 +165,53 @@ func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(t *testing.T) {
require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref") require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref")
brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read) brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read)
} }
// hyperguild#158: BackfillRefs read a regular concept page, a symlink could
// be swapped in, and os.WriteFile then followed it and overwrote the link's
// target. The target is not restored between rounds here, so any
// write-through shows.
func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "s.md"), []byte("---\ntitle: S\n---\n[[c|C]]\n"), 0o644))
page := filepath.Join(brain, "wiki", "concepts", "c.md")
target := filepath.Join(brain, "raw", "secret.md")
require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644))
inside := "---\ntitle: C\n---\nbody\n"
require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
brainfstest.StartSwap(t, page, inside, target, false)
ok := 0
for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
if _, err := BackfillRefs(context.Background(), brain); err == nil {
ok++
}
}
got, _ := os.ReadFile(target)
assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink")
assert.Positive(t, ok, "positive control: some backfills ran")
}
// hyperguild#158: buildAndWrite wrote each page with os.WriteFile by path, so
// a page read as regular and swapped for a symlink before the write was
// written through. The target is not restored between rounds, so any
// write-through shows.
func TestBuildAndWriteNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
brain := t.TempDir()
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "wiki/entities", "raw")
page := filepath.Join(brain, "wiki", "concepts", "zq-concept.md")
target := filepath.Join(brain, "raw", "secret.md")
require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644))
inside := "---\ntitle: Zq Concept\n---\nbody\n"
require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
pages := []RawPage{{Title: "Zq Concept", Type: "concept", Content: "## About\n\nmore\n"}}
brainfstest.StartSwap(t, page, inside, target, false)
ok := 0
for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
if _, err := RunRaw(context.Background(), brain, "zq source", pages, false); err == nil {
ok++
}
}
got, _ := os.ReadFile(target)
assert.Equal(t, "untouched", string(got), "buildAndWrite wrote through a swapped-in symlink")
assert.Positive(t, ok, "positive control: some runs returned without error")
}
@@ -1,138 +0,0 @@
package pipeline
import (
"context"
"os"
"path/filepath"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func wingCfg(wing string, pages ...RawPage) Config {
return Config{
Wing: wing,
Schema: "# Schema",
Complete: func(context.Context, string, string) (string, error) {
return mustJSON(pages), nil
},
}
}
func sourceAndEntity() []RawPage {
return []RawPage{
{Title: "Dark Factory Talk", Type: "source", Subtype: "video", Domain: "ai",
Content: "## Summary\n\nA talk.\n\n## Entities Mentioned\n\n[[Jev]]\n"},
{Title: "Jev", Type: "entity", Subtype: "tool", Domain: "ai",
Content: "## Definition\n\nA decision model.\n\n## Sources\n"},
}
}
func exists(dir, rel string) bool {
_, err := os.Stat(filepath.Join(dir, rel))
return err == nil
}
func read(t *testing.T, dir, rel string) string {
t.Helper()
b, err := os.ReadFile(filepath.Join(dir, rel))
require.NoError(t, err)
return string(b)
}
func TestRun_WithAWingWritesIntoThatWingsHalls(t *testing.T) {
dir := t.TempDir()
_, err := Run(context.Background(), wingCfg("radar", sourceAndEntity()...), dir, "body of the talk", "dark-factory-talk", false)
require.NoError(t, err)
src := read(t, dir, "wiki/radar/sources/dark-factory-talk.md")
assert.Contains(t, src, "wing: radar")
assert.Contains(t, src, "hall: sources")
ent := read(t, dir, "wiki/radar/facts/jev.md")
assert.Contains(t, ent, "wing: radar")
assert.Contains(t, ent, "hall: facts")
assert.Contains(t, ent, "kind: entity")
}
func TestRun_WithAWingDoesNotTouchTheRetiredLayout(t *testing.T) {
dir := t.TempDir()
_, err := Run(context.Background(), wingCfg("radar", sourceAndEntity()...), dir, "body of the talk", "dark-factory-talk", false)
require.NoError(t, err)
for _, retired := range []string{"wiki/sources", "wiki/concepts", "wiki/entities"} {
assert.False(t, exists(dir, retired), "%s is the retired layout and must not be written", retired)
}
}
func TestRun_WithAWingTheSourceLinksBackFromTheEntity(t *testing.T) {
dir := t.TempDir()
_, err := Run(context.Background(), wingCfg("radar", sourceAndEntity()...), dir, "body of the talk", "dark-factory-talk", false)
require.NoError(t, err)
assert.Contains(t, read(t, dir, "wiki/radar/facts/jev.md"), "[[dark-factory-talk|Dark Factory Talk]]")
}
func TestRun_WithAWingASecondSourceReusesTheSameEntityPage(t *testing.T) {
dir := t.TempDir()
_, err := Run(context.Background(), wingCfg("radar", sourceAndEntity()...), dir, "first body", "talk-one", false)
require.NoError(t, err)
again := []RawPage{
{Title: "Second Talk", Type: "source", Subtype: "video", Content: "## Summary\n\nMore.\n\n## Entities Mentioned\n\n[[The Jev]]\n"},
{Title: "The Jev", Type: "entity", Subtype: "tool", Content: "## Definition\n\nStill a decision model.\n\n## Sources\n"},
}
_, err = Run(context.Background(), wingCfg("radar", again...), dir, "second body", "talk-two", false)
require.NoError(t, err)
files, err := filepath.Glob(filepath.Join(dir, "wiki/radar/facts/*.md"))
require.NoError(t, err)
assert.Len(t, files, 1, "an entity matched by title (leading article ignored) is merged, not duplicated")
assert.Contains(t, read(t, dir, "wiki/radar/facts/jev.md"), "talk-two")
}
func TestRun_WithAWingNeverMergesIntoAnotherWingsPage(t *testing.T) {
dir := t.TempDir()
other := "---\ntitle: Jev\nwing: client-x\nhall: facts\nkind: entity\n---\n\n## Definition\n\nPrivate to another wing.\n"
require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki/client-x/facts"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "wiki/client-x/facts/jev.md"), []byte(other), 0o644))
_, err := Run(context.Background(), wingCfg("radar", sourceAndEntity()...), dir, "body", "dark-factory-talk", false)
require.NoError(t, err)
assert.Equal(t, other, read(t, dir, "wiki/client-x/facts/jev.md"), "another wing's page is never read into or rewritten")
assert.True(t, exists(dir, "wiki/radar/facts/jev.md"))
}
func TestRun_WithAWingRejectsANameThatIsNotAPlainWing(t *testing.T) {
for _, bad := range []string{"../escape", "a/b", "Radar", "has space", "."} {
t.Run(bad, func(t *testing.T) {
dir := t.TempDir()
_, err := Run(context.Background(), wingCfg(bad, sourceAndEntity()...), dir, "body", "talk", false)
require.Error(t, err)
assert.Len(t, listAll(t, dir), 1, "nothing is written for a rejected wing (only the temp dir itself exists)")
})
}
}
func listAll(t *testing.T, dir string) []string {
t.Helper()
var out []string
_ = filepath.WalkDir(dir, func(p string, _ os.DirEntry, _ error) error { out = append(out, p); return nil })
return out
}
// The dangerous path is the back-reference: a source that links [[Jev]] without
// proposing a Jev page makes the pipeline go and edit an existing Jev page. That
// lookup must be confined to the target wing.
func TestRun_WithAWingASourceLinkingToAnotherWingsEntityDoesNotEditIt(t *testing.T) {
dir := t.TempDir()
other := "---\ntitle: Jev\nwing: client-x\nhall: facts\nkind: entity\n---\n\n## Definition\n\nPrivate to another wing.\n\n## Sources\n"
require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki/client-x/facts"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "wiki/client-x/facts/jev.md"), []byte(other), 0o644))
onlySource := sourceAndEntity()[:1]
_, err := Run(context.Background(), wingCfg("radar", onlySource...), dir, "body", "dark-factory-talk", false)
require.NoError(t, err)
assert.Equal(t, other, read(t, dir, "wiki/client-x/facts/jev.md"), "no back-reference may be written into another wing")
}
@@ -381,19 +381,18 @@ func TestWatchQueueIsBoundedDuringAnOutage(t *testing.T) {
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond) require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
} }
n.down.Store(false) n.down.Store(false)
// Synchronise on the summary being delivered (hyperguild#193: a fixed require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 5*time.Millisecond)
// sleep flaked under load); once it is, the queue has been flushed. time.Sleep(100 * time.Millisecond)
require.Eventually(t, func() bool { assert.LessOrEqual(t, n.count(), maxPending+1, "queued alerts plus at most one suppression summary")
n.mu.Lock() n.mu.Lock()
defer n.mu.Unlock() defer n.mu.Unlock()
found := false
for _, m := range n.msgs { for _, m := range n.msgs {
if strings.Contains(m, "suppressed") { if strings.Contains(m, "suppressed") {
return true found = true
} }
} }
return false assert.True(t, found, "suppressed alerts are counted in a summary")
}, 10*time.Second, 5*time.Millisecond, "suppressed alerts are counted in a summary")
assert.LessOrEqual(t, n.count(), maxPending+1, "queued alerts plus at most one suppression summary")
} }
// Within the bound, every alert queued during an outage is delivered after // Within the bound, every alert queued during an outage is delivered after
@@ -412,21 +411,11 @@ func TestWatchDeliversEveryQueuedAlertAfterAShortOutage(t *testing.T) {
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond) require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
} }
n.down.Store(false) n.down.Store(false)
require.Eventually(t, func() bool { return n.count() >= 3 }, 10*time.Second, 5*time.Millisecond) require.Eventually(t, func() bool { return n.count() == 3 }, 2*time.Second, 5*time.Millisecond)
n.mu.Lock() n.mu.Lock()
defer n.mu.Unlock() defer n.mu.Unlock()
// Each removal delivered exactly once, in any order: the outage ends for i, name := range []string{"qa", "qb", "qc"} {
// asynchronously, possibly mid-flush, so a later item can send before an assert.Contains(t, n.msgs[i], strconv.Quote(name))
// earlier one (hyperguild#193). Order is not what this test is about.
assert.Len(t, n.msgs, 3)
for _, name := range []string{"qa", "qb", "qc"} {
hits := 0
for _, m := range n.msgs {
if strings.Contains(m, strconv.Quote(name)) {
hits++
}
}
assert.Equal(t, 1, hits, "%s delivered exactly once", name)
} }
} }
@@ -468,30 +457,29 @@ func TestWatchAccountsForEveryRemovalDuringAnOutage(t *testing.T) {
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 1*time.Millisecond) require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 1*time.Millisecond)
} }
n.down.Store(false) n.down.Store(false)
// Synchronise on the record being complete, not on a fixed sleep after require.Eventually(t, func() bool {
// the first summary (hyperguild#193): sum every "removed N" alert and
// every suppression count until they account for all plants.
account := func() (delivered, counted int) {
n.mu.Lock() n.mu.Lock()
defer n.mu.Unlock() defer n.mu.Unlock()
for _, m := range n.msgs { for _, m := range n.msgs {
var k int if strings.Contains(m, "suppressed") {
if i := strings.Index(m, "removed "); i >= 0 && strings.Contains(m, "symlink(s)") { return true
if _, err := fmt.Sscanf(m[i+len("removed "):], "%d", &k); err == nil {
delivered += k
} }
} }
return false
}, 2*time.Second, 5*time.Millisecond)
time.Sleep(50 * time.Millisecond)
n.mu.Lock()
defer n.mu.Unlock()
delivered, counted := 0, 0
for _, m := range n.msgs {
if strings.Contains(m, "removed 1 symlink(s)") {
delivered++
}
if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") { if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
if _, err := fmt.Sscanf(m[i+3:], "%d", &k); err == nil { _, _ = fmt.Sscanf(m[i+3:], "%d", &counted)
counted += k
} }
} }
} assert.Equal(t, plants, delivered+counted, "delivered %d + counted %d must equal %d removals", delivered, counted, plants)
return delivered, counted
}
require.Eventually(t, func() bool { d, c := account(); return d+c >= plants }, 10*time.Second, 5*time.Millisecond)
d, c := account()
assert.Equal(t, plants, d+c, "delivered %d + counted %d must equal %d removals", d, c, plants)
} }
// A poison message is still dropped (others send while it fails), and the // A poison message is still dropped (others send while it fails), and the
-44
View File
@@ -1,44 +0,0 @@
package watcher
import (
"errors"
"fmt"
"io/fs"
"path/filepath"
"strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
)
// Kept out of watcher.go on purpose: that file is covered by the raw-reader
// guard (internal/extract/callsites_test.go), which forbids any .Open selector
// there. This function reads nothing from raw/; it creates a file under the
// brain root, which refuses symlinked directories and never follows a final
// symlink.
// createParked writes data to raw/needs-review/<filename> through the brain
// root, never replacing a parked note: on a name clash it tries -2, -3, ...
// A needs-review that is a symlink, or any non-directory, is refused.
func createParked(brainDir, filename string, data []byte) error {
bfs, err := brainfs.Open(brainDir)
if err != nil {
return err
}
defer func() { _ = bfs.Close() }()
ext := filepath.Ext(filename)
stem := strings.TrimSuffix(filename, ext)
for n := 1; n < 10000; n++ {
name := filename
if n > 1 {
name = fmt.Sprintf("%s-%d%s", stem, n, ext)
}
err := bfs.CreateRegular("raw/needs-review/"+name, data)
if err == nil {
return nil
}
if !errors.Is(err, fs.ErrExist) {
return err
}
}
return fmt.Errorf("no free name for %q in raw/needs-review", filename)
}
+72 -219
View File
@@ -3,6 +3,7 @@ package watcher
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"io" "io"
"log/slog" "log/slog"
@@ -13,10 +14,9 @@ import (
"time" "time"
"unicode" "unicode"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/extract" "github.com/mathiasbq/hyperguild/ingestion/internal/extract"
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
) )
// Config holds watcher configuration. // Config holds watcher configuration.
@@ -24,24 +24,11 @@ type Config struct {
BrainDir string BrainDir string
Interval time.Duration Interval time.Duration
Pipeline pipeline.Config Pipeline pipeline.Config
// stuck remembers notes that were ingested but could neither be removed
// from raw/ nor marked (e.g. a read-only raw/), so later polls skip them
// instead of ingesting them again. In memory only: a restart retries once.
stuck *sync.Map
// Resolver, when set, decides each note's wing (hyperguild#205): explicit
// subfolder/frontmatter, else classifier + scorer. A note it cannot place
// is parked in raw/needs-review/ for a human. Nil keeps the retired layout.
Resolver *wing.Resolver
} }
// Start launches the watcher in a background goroutine. // Start launches the watcher in a background goroutine.
// It returns immediately. The watcher stops when ctx is cancelled. // It returns immediately. The watcher stops when ctx is cancelled.
func Start(ctx context.Context, cfg Config) { func Start(ctx context.Context, cfg Config) {
if cfg.stuck == nil {
cfg.stuck = &sync.Map{}
}
go func() { go func() {
ticker := time.NewTicker(cfg.Interval) ticker := time.NewTicker(cfg.Interval)
defer ticker.Stop() defer ticker.Stop()
@@ -101,7 +88,7 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
// path that treats raw/ as "awaiting processing" — this is that path. // path that treats raw/ as "awaiting processing" — this is that path.
if d.IsDir() { if d.IsDir() {
name := d.Name() name := d.Name()
if name == "processed" || name == "failed" || name == "rejected" || name == "needs-review" { if name == "processed" || name == "failed" || name == "rejected" {
return filepath.SkipDir return filepath.SkipDir
} }
return nil return nil
@@ -124,21 +111,29 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
return nil return nil
} }
if cfg.stuck != nil {
if _, skip := cfg.stuck.Load(path); skip {
return nil
}
}
// Skip files that have already been processed or permanently failed. // Skip files that have already been processed or permanently failed.
if _, err := os.Stat(path + ".processed"); err == nil { // Any entry at the marker path counts (Lstat, not Stat): a dangling
// marker symlink made os.Stat report nothing and the note was
// reprocessed every interval (hyperguild#158).
if _, err := os.Lstat(path + ".processed"); err == nil {
return nil return nil
} }
if _, err := os.Stat(path + ".failed"); err == nil { if _, err := os.Lstat(path + ".failed"); err == nil {
return nil
}
// A note whose archive copy or marker was refused is not retried
// until it changes: a refusal has no marker to stop the next round,
// and each round can cost an LLM call (hyperguild#158).
info, ierr := d.Info()
if ierr == nil && refusedUnchanged(path, info) {
return nil return nil
} }
if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil { if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil {
if ierr == nil && errors.Is(err, brainfs.ErrNotRegular) {
rememberRefused(path, info)
slog.Error("watcher: note refused; not retried until it changes", "file", filepath.Base(path), "error", err)
}
errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err)) errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err))
} }
return nil return nil
@@ -149,12 +144,9 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
return errs return errs
} }
// processFile reads a file, calls pipeline.Run, and archives it into processed/ // processFile reads a file, calls pipeline.Run, copies it to processed/ or failed/,
// or failed/. The original is then removed from raw/ so that raw/ only ever // and writes a marker file next to the original so the watcher skips it next poll.
// holds unprocessed input (inbox zero). The removal syncs to Syncthing peers; // The original file is never deleted, keeping Syncthing-connected vaults (e.g. Obsidian) intact.
// the archive copy under raw/processed/ or raw/failed/ is what they keep. A
// marker is written next to the original only if the removal fails, so a
// stuck file is not re-ingested every poll.
func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, date string) error { func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, date string) error {
filename := filepath.Base(path) filename := filepath.Base(path)
source := deriveSource(filename) source := deriveSource(filename)
@@ -163,40 +155,24 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
return fmt.Errorf("relative path: %w", err) return fmt.Errorf("relative path: %w", err)
} }
before, err := root.Lstat(rel)
if err != nil {
return fmt.Errorf("stat before ingest: %w", err)
}
content, err := extract.TextIn(root, rel) content, err := extract.TextIn(root, rel)
if err != nil { if err != nil {
return fmt.Errorf("extract text: %w", err) return fmt.Errorf("extract text: %w", err)
} }
pcfg := cfg.Pipeline _, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false)
if cfg.Resolver != nil {
d := cfg.Resolver.Resolve(ctx, wing.Note{Dir: topDir(rel), FrontmatterWing: frontmatterWing(content), Text: content})
if d.Source == wing.Exception {
return parkForReview(cfg, root, rel, path, filename, content, d, date, before)
}
pcfg.Wing = d.Wing
slog.Info("watcher: wing resolved", "file", filename, "wing", d.Wing, "how", d.Source, "reason", d.Reason)
}
_, runErr := pipeline.Run(ctx, pcfg, cfg.BrainDir, content, source, false)
if runErr != nil { if runErr != nil {
// Copy to failed/ and leave a .failed marker so we don't retry. // Copy to failed/ and leave a .failed marker so we don't retry. Both
failedDir := filepath.Join(cfg.BrainDir, "raw", "failed") // written through the brain root (hyperguild#158).
if mkErr := os.MkdirAll(failedDir, 0o755); mkErr != nil { if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil {
return fmt.Errorf("mkdir failed dir: %w", mkErr)
}
dest := filepath.Join(failedDir, filename)
if cpErr := copyFromRoot(root, rel, dest); cpErr != nil {
return fmt.Errorf("copy to failed: %w", cpErr) return fmt.Errorf("copy to failed: %w", cpErr)
} }
if changedSince(root, rel, before) { if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil {
slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename) if errors.Is(mkErr, brainfs.ErrNotRegular) {
return nil return fmt.Errorf("write .failed marker: %w", mkErr)
}
slog.Error("watcher: failed to write .failed marker", "error", mkErr)
} }
retire(cfg, root, rel, path, ".failed", runErr.Error())
slog.Warn("watcher: file failed", "file", filename, "error", runErr) slog.Warn("watcher: file failed", "file", filename, "error", runErr)
@@ -208,57 +184,52 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
} }
// Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest. // Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest.
processedDir := filepath.Join(cfg.BrainDir, "raw", "processed", date) // Written through the brain root (hyperguild#158).
if err := os.MkdirAll(processedDir, 0o755); err != nil { dest := "raw/processed/" + date + "/" + filename
return fmt.Errorf("mkdir processed dir: %w", err) if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil {
}
dest := filepath.Join(processedDir, filename)
if _, err := os.Stat(dest); err == nil {
// Archive copy already exists; append timestamp to avoid overwriting. // Archive copy already exists; append timestamp to avoid overwriting.
ext := filepath.Ext(filename) ext := filepath.Ext(filename)
base := strings.TrimSuffix(filename, ext) base := strings.TrimSuffix(filename, ext)
dest = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext) dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext
} }
if err := copyFromRoot(root, rel, dest); err != nil { if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil {
return fmt.Errorf("copy to processed: %w", err) return fmt.Errorf("copy to processed: %w", err)
} }
if changedSince(root, rel, before) { if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil {
slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename) if errors.Is(err, brainfs.ErrNotRegular) {
return nil return fmt.Errorf("write .processed marker: %w", err)
}
slog.Error("watcher: failed to write .processed marker", "error", err)
} }
retire(cfg, root, rel, path, ".processed", date)
slog.Info("watcher: file processed", "file", filename, "source", source) slog.Info("watcher: file processed", "file", filename, "source", source)
return nil return nil
} }
// changedSince reports whether the note at rel differs (size or mtime) from // refused remembers notes whose archive copy or marker write was refused,
// before. A note replaced mid-ingest (Syncthing) was extracted from older // by path, with the size and mtime they had, for the life of the process.
// content than it now holds, so it must not be removed. Limits: a same-size var refused sync.Map
// rewrite with a restored mtime is not detected (the new bytes then survive
// only in the archive copy), nor a change between this check and Remove. type refusedStamp struct {
func changedSince(root *os.Root, rel string, before os.FileInfo) bool { size int64
now, err := root.Lstat(rel) mtime time.Time
if err != nil {
return true
}
return now.Size() != before.Size() || !now.ModTime().Equal(before.ModTime())
} }
// retire removes an archived original from raw/ (inbox zero). If the removal func rememberRefused(path string, info os.FileInfo) {
// fails it falls back to a marker file so the watcher still skips the file. refused.Store(path, refusedStamp{size: info.Size(), mtime: info.ModTime()})
func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) { }
rmErr := root.Remove(rel)
if rmErr == nil { func refusedUnchanged(path string, info os.FileInfo) bool {
return v, ok := refused.Load(path)
} if !ok {
slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr) return false
if err := os.WriteFile(path+markerSuffix, []byte(markerBody), 0o644); err != nil {
slog.Error("watcher: failed to write marker; skipping this note until restart", "file", rel, "error", err)
if cfg.stuck != nil {
cfg.stuck.Store(path, struct{}{})
} }
st := v.(refusedStamp)
if st.size == info.Size() && st.mtime.Equal(info.ModTime()) {
return true
} }
refused.Delete(path)
return false
} }
// deriveSource turns a filename into a human-readable source name. // deriveSource turns a filename into a human-readable source name.
@@ -285,23 +256,22 @@ func appendWatcherLog(brainDir, filename string, runErr error, date string) erro
entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n", entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
date, filename, runErr.Error()) date, filename, runErr.Error())
logPath := filepath.Join(brainDir, "log.md") if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil {
f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if err != nil {
return fmt.Errorf("open log: %w", err)
}
if _, err = f.WriteString(entry); err != nil {
f.Close() //nolint:errcheck
return fmt.Errorf("write log: %w", err) return fmt.Errorf("write log: %w", err)
} }
return f.Close() return nil
}
// putBrainFile creates or replaces a brain-relative file through the brain
// root, never through a symlinked directory or file (hyperguild#158).
func putBrainFile(brainDir, rel string, data []byte) error {
return brainfs.PutFile(brainDir, rel, data)
} }
// copyFromRoot archives a raw/ file read through the root, refusing anything // copyFromRoot archives a raw/ file read through the root, refusing anything
// that is not a regular file, so a swapped-in symlink cannot copy an outside // that is not a regular file, so a swapped-in symlink cannot copy an outside
// file into raw/processed/ or raw/failed/ (hyperguild#153). // file into raw/processed/ or raw/failed/ (hyperguild#153).
func copyFromRoot(root *os.Root, rel, dst string) error { func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error {
fi, err := root.Lstat(rel) fi, err := root.Lstat(rel)
if err != nil { if err != nil {
return err return err
@@ -314,126 +284,9 @@ func copyFromRoot(root *os.Root, rel, dst string) error {
return err return err
} }
defer func() { _ = in.Close() }() defer func() { _ = in.Close() }()
out, err := os.Create(dst) data, err := io.ReadAll(in)
if err != nil { if err != nil {
return err return err
} }
if _, err := io.Copy(out, in); err != nil { return putBrainFile(brainDir, dstRel, data)
_ = out.Close()
return err
}
return out.Close()
}
// topDir is the first directory of a raw/-relative path, "" for a top-level file.
func topDir(rel string) string {
if i := strings.IndexByte(filepath.ToSlash(rel), '/'); i > 0 {
return filepath.ToSlash(rel)[:i]
}
return ""
}
// frontmatterWing reads `wing:` from a note's leading frontmatter, "" if none.
func frontmatterWing(content string) string {
content = strings.ReplaceAll(content, "\r\n", "\n")
if !strings.HasPrefix(content, "---\n") {
return ""
}
block, _, _ := strings.Cut(content[4:], "\n---")
for _, line := range strings.Split(block, "\n") {
if key, val, ok := strings.Cut(line, ":"); ok && strings.TrimSpace(key) == "wing" {
return strings.Trim(strings.TrimSpace(val), `"'`)
}
}
return ""
}
// parkForReview moves a note the resolver could not place into
// raw/needs-review/, recording the proposal in its frontmatter so a human can
// approve it by moving it into raw/<wing>/. Nothing is extracted or written to
// the wiki.
//
// The parked copy is created through the brain root (real directories only,
// O_EXCL, never overwriting) and is the only copy once the original is retired,
// so the original is removed only after that write succeeds and only if it has
// not changed since it was read: a newer version delivered while the resolver
// ran stays in raw/ for the next poll.
func parkForReview(cfg Config, root *os.Root, rel, path, filename, content string, d wing.Decision, date string, before os.FileInfo) error {
var data []byte
if strings.ToLower(filepath.Ext(filename)) == ".pdf" {
// Binary: park the bytes untouched; the proposal is in log.md only.
in, _, err := extract.OpenRegular(root, rel)
if err != nil {
return fmt.Errorf("park pdf: %w", err)
}
data, err = io.ReadAll(in)
_ = in.Close()
if err != nil {
return fmt.Errorf("park pdf: %w", err)
}
} else {
data = []byte(withProposal(content, d))
}
if err := createParked(cfg.BrainDir, filename, data); err != nil {
return fmt.Errorf("park note: %w", err)
}
if err := appendWatcherLog(cfg.BrainDir, filename, fmt.Errorf("needs review (proposed wing %q): %s", d.Wing, d.Reason), date); err != nil {
slog.Error("watcher: failed to write log entry", "error", err)
}
slog.Info("watcher: parked for review", "file", filename, "proposed_wing", d.Wing, "reason", d.Reason)
if changedSince(root, rel, before) {
slog.Warn("watcher: note changed while it was being resolved; leaving the newer version in raw/", "file", filename)
return nil
}
retire(cfg, root, rel, path, ".processed", date)
return nil
}
// proposalKeys are the frontmatter keys the resolver owns in a parked note.
var proposalKeys = map[string]bool{"proposed_wing": true, "proposal_confidence": true, "proposal_reason": true}
// withProposal adds the resolver's proposal to a note's frontmatter, creating
// the block when the note has none. Every value is on one line: the wing only
// if it is a plain wing name (else it appears in the reason alone), strings
// Go-quoted. Any proposal keys the note already carried are dropped first, so a
// note cannot pre-empt the resolver's (frontmatter reads are first-wins).
// Line endings are normalised to LF so a CRLF note keeps one valid block.
func withProposal(content string, d wing.Decision) string {
content = strings.ReplaceAll(content, "\r\n", "\n")
reason := d.Reason
var kv strings.Builder
if d.Wing != "" && brain.Sanitise(d.Wing) == d.Wing {
fmt.Fprintf(&kv, "proposed_wing: %s\n", d.Wing)
} else if d.Wing != "" {
reason = fmt.Sprintf("proposed wing %q is not a plain wing name; %s", d.Wing, reason)
}
fmt.Fprintf(&kv, "proposal_confidence: %.2f\nproposal_reason: %q\n", d.Confidence, reason)
if !strings.HasPrefix(content, "---\n") {
return "---\n" + kv.String() + "---\n" + content
}
block, rest, found := cutFence(content[4:])
if !found {
return "---\n" + kv.String() + "---\n" + content
}
var kept []string
for _, line := range strings.Split(block, "\n") {
if key, _, ok := strings.Cut(line, ":"); ok && proposalKeys[strings.TrimSpace(key)] && !strings.HasPrefix(line, " ") {
continue
}
kept = append(kept, line)
}
return "---\n" + kv.String() + strings.Join(kept, "\n") + "\n---" + rest
}
// cutFence splits s at the first closing fence: a line that is exactly "---".
// "----" or "---x" is not a fence, matching how the frontmatter readers parse.
func cutFence(s string) (block, rest string, found bool) {
if i := strings.Index(s, "\n---\n"); i >= 0 {
return s[:i], s[i+4:], true
}
if strings.HasSuffix(s, "\n---") {
return s[:len(s)-4], "", true
}
return "", "", false
} }
+10 -90
View File
@@ -89,12 +89,13 @@ func TestStart_ProcessesFile(t *testing.T) {
} }
require.True(t, found, "file should be copied to processed/") require.True(t, found, "file should be copied to processed/")
// Inbox zero: once archived, the original leaves raw/ and no marker is // Original file should still exist (copy, not move — keeps Obsidian vault intact).
// left behind, so what remains in raw/ is exactly the unprocessed set.
_, err := os.Stat(rawFile) _, err := os.Stat(rawFile)
assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once archived") assert.NoError(t, err, "original file should remain in raw/")
// A .processed marker should exist next to the original.
_, err = os.Stat(rawFile + ".processed") _, err = os.Stat(rawFile + ".processed")
assert.ErrorIs(t, err, os.ErrNotExist, "no .processed marker: the archive copy is the record") assert.NoError(t, err, ".processed marker should be written")
// Wiki page should exist. // Wiki page should exist.
wikiPath := filepath.Join(brainDir, "wiki", "sources", "shape-up-book.md") wikiPath := filepath.Join(brainDir, "wiki", "sources", "shape-up-book.md")
@@ -141,11 +142,13 @@ func TestStart_MovesToFailedOnError(t *testing.T) {
} }
require.True(t, found, "file should be copied to failed/") require.True(t, found, "file should be copied to failed/")
// Inbox zero: a quarantined original leaves raw/ too, no marker beside it. // Original file should still exist (copy, not move — keeps Obsidian vault intact).
_, err := os.Stat(rawFile) _, err := os.Stat(rawFile)
assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once quarantined") assert.NoError(t, err, "original file should remain in raw/")
// A .failed marker should exist next to the original.
_, err = os.Stat(rawFile + ".failed") _, err = os.Stat(rawFile + ".failed")
assert.ErrorIs(t, err, os.ErrNotExist, "no .failed marker: failed/ is the record") assert.NoError(t, err, ".failed marker should be written")
// log.md should contain a watcher error entry. // log.md should contain a watcher error entry.
logContent, err := os.ReadFile(filepath.Join(brainDir, "log.md")) logContent, err := os.ReadFile(filepath.Join(brainDir, "log.md"))
@@ -387,86 +390,3 @@ func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) {
t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load()) t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load())
} }
} }
// A note whose removal AND marker write both fail (read-only raw/) must not be
// re-ingested on every poll: each pass would write another archive copy.
func TestProcessDir_DoesNotReingestANoteItCannotRetire(t *testing.T) {
brainDir := setupBrainDir(t)
rawDir := filepath.Join(brainDir, "raw")
require.NoError(t, os.WriteFile(filepath.Join(rawDir, "stuck.md"), []byte("Content."), 0o644))
// processed/ must exist and stay writable; only raw/ itself is locked.
require.NoError(t, os.MkdirAll(filepath.Join(rawDir, "processed"), 0o755))
require.NoError(t, os.Chmod(rawDir, 0o555))
t.Cleanup(func() { _ = os.Chmod(rawDir, 0o755) })
var calls atomic.Int32
cfg := Config{
BrainDir: brainDir,
Pipeline: pipeline.Config{
Complete: func(ctx context.Context, s, u string) (string, error) {
calls.Add(1)
return successComplete(pipeline.RawPage{Title: "Stuck", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u)
},
Schema: "# Schema",
},
stuck: &sync.Map{},
}
for i := 0; i < 3; i++ {
processDir(context.Background(), cfg, "2026-10-19")
}
assert.Equal(t, int32(1), calls.Load(), "a note that cannot be retired is ingested once, not once per poll")
}
// If the archive copy cannot be written, the original is the only copy left:
// it must stay in raw/.
func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) {
brainDir := setupBrainDir(t)
rawDir := filepath.Join(brainDir, "raw")
note := filepath.Join(rawDir, "keep-me.md")
require.NoError(t, os.WriteFile(note, []byte("Content."), 0o644))
// The dated archive dir exists but is read-only: MkdirAll succeeds, the
// copy itself (os.Create) fails.
dated := filepath.Join(rawDir, "processed", "2026-10-19")
require.NoError(t, os.MkdirAll(dated, 0o755))
require.NoError(t, os.Chmod(dated, 0o555))
t.Cleanup(func() { _ = os.Chmod(dated, 0o755) })
cfg := Config{
BrainDir: brainDir,
Pipeline: pipeline.Config{
Complete: successComplete(pipeline.RawPage{Title: "Keep Me", Type: "source", Subtype: "article", Content: "x\n"}),
Schema: "# Schema",
},
}
errs := processDir(context.Background(), cfg, "2026-10-19")
assert.NotEmpty(t, errs, "the failed archive is reported")
_, err := os.Stat(note)
assert.NoError(t, err, "original stays in raw/ when it could not be archived")
}
// If the note is replaced (e.g. by Syncthing) while it is being ingested, the
// wiki page was extracted from the old content; removing the original would
// leave the new content only in the archive and never re-ingest it. Observes
// size/mtime change only: a same-size rewrite with a restored mtime is not caught.
func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) {
brainDir := setupBrainDir(t)
note := filepath.Join(brainDir, "raw", "edited.md")
require.NoError(t, os.WriteFile(note, []byte("old"), 0o644))
cfg := Config{
BrainDir: brainDir,
Pipeline: pipeline.Config{
Complete: func(ctx context.Context, s, u string) (string, error) {
require.NoError(t, os.WriteFile(note, []byte("newer and longer content"), 0o644))
return successComplete(pipeline.RawPage{Title: "Edited", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u)
},
Schema: "# Schema",
},
}
processDir(context.Background(), cfg, "2026-10-19")
got, err := os.ReadFile(note)
require.NoError(t, err, "a note that changed during ingest must stay in raw/")
assert.Equal(t, "newer and longer content", string(got))
}
@@ -1,331 +0,0 @@
package watcher
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"sync/atomic"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
type stubClassifier struct {
s wing.Suggestion
err error
calls atomic.Int32
}
func (c *stubClassifier) Classify(context.Context, string, []string) (wing.Suggestion, error) {
c.calls.Add(1)
return c.s, c.err
}
type stubScorer map[string]float64
func (s stubScorer) Score(string) map[string]float64 { return s }
func wingResolver(c wing.Classifier, s wing.Scorer) *wing.Resolver {
return &wing.Resolver{
Known: []string{"radar", "homelab", "client-x"},
Eligible: func(w string) bool { return w != "client-x" },
Classifier: c,
Scorer: s,
Threshold: 0.8,
}
}
func wingTestCfg(t *testing.T, r *wing.Resolver, llmCalls *atomic.Int32) (Config, string) {
t.Helper()
brainDir := setupBrainDir(t)
return Config{
BrainDir: brainDir,
Resolver: r,
Pipeline: pipeline.Config{
Schema: "# Schema",
Complete: func(ctx context.Context, s, u string) (string, error) {
llmCalls.Add(1)
return successComplete(pipeline.RawPage{Title: "Talk", Type: "source", Subtype: "video", Content: "## Summary\n\nx\n"})(ctx, s, u)
},
},
}, brainDir
}
func put(t *testing.T, brainDir, rel, body string) {
t.Helper()
full := filepath.Join(brainDir, "raw", rel)
require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755))
require.NoError(t, os.WriteFile(full, []byte(body), 0o644))
}
func has(brainDir, rel string) bool {
_, err := os.Stat(filepath.Join(brainDir, rel))
return err == nil
}
func TestProcessDir_ASubfolderNamesTheWing(t *testing.T) {
var llmCalls atomic.Int32
clf := &stubClassifier{}
cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{}), &llmCalls)
put(t, brainDir, "radar/dark-factory.md", "A talk about factories.")
errs := processDir(context.Background(), cfg, "2026-10-20")
require.Empty(t, errs)
assert.True(t, has(brainDir, "wiki/radar/sources/dark-factory.md"), "lands in the named wing")
assert.False(t, has(brainDir, "raw/radar/dark-factory.md"), "inbox zero holds for subfolders too")
assert.Zero(t, clf.calls.Load(), "an explicit wing costs no classification")
}
func TestProcessDir_ANoteTheResolverCannotPlaceIsParkedNotIngested(t *testing.T) {
var llmCalls atomic.Int32
clf := &stubClassifier{s: wing.Suggestion{Wing: "radar", Confidence: 0.5, Reason: "maybe"}}
cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{"radar": 3}), &llmCalls)
put(t, brainDir, "unsure.md", "Some note.")
errs := processDir(context.Background(), cfg, "2026-10-20")
require.Empty(t, errs)
assert.Zero(t, llmCalls.Load(), "no extraction runs for an unplaced note")
assert.False(t, has(brainDir, "raw/unsure.md"), "leaves the top of raw/")
parked, err := os.ReadFile(filepath.Join(brainDir, "raw", "needs-review", "unsure.md"))
require.NoError(t, err)
assert.Contains(t, string(parked), "proposed_wing: radar")
assert.Contains(t, string(parked), "below threshold")
assert.Contains(t, string(parked), "Some note.", "the note body is preserved")
}
func TestProcessDir_ANoteInAClientWingFolderIsParked(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{}, stubScorer{}), &llmCalls)
put(t, brainDir, "client-x/secret.md", "Client material.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
assert.Zero(t, llmCalls.Load())
assert.False(t, has(brainDir, "wiki/client-x"), "nothing is ever written into an ineligible wing")
assert.True(t, has(brainDir, "raw/needs-review/secret.md"))
}
func TestProcessDir_ClassifierFailureParksTheNote(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{err: errors.New("llm down")}, stubScorer{}), &llmCalls)
put(t, brainDir, "n.md", "Note.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
assert.Zero(t, llmCalls.Load())
assert.True(t, has(brainDir, "raw/needs-review/n.md"))
}
func TestProcessDir_ParkedNotesAreNotReprocessed(t *testing.T) {
var llmCalls atomic.Int32
clf := &stubClassifier{s: wing.Suggestion{Wing: "radar", Confidence: 0.99}}
cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{"radar": 3}), &llmCalls)
put(t, brainDir, "needs-review/waiting.md", "---\nproposed_wing: radar\n---\nA parked note.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
assert.Zero(t, clf.calls.Load(), "needs-review/ is a human queue, like rejected/")
assert.True(t, has(brainDir, "raw/needs-review/waiting.md"))
}
func TestProcessDir_WithoutAResolverTheRetiredLayoutIsUnchanged(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, nil, &llmCalls)
put(t, brainDir, "plain.md", "Plain note.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
assert.True(t, has(brainDir, "wiki/sources/plain.md"))
}
// The reason comes from an LLM and ends up in frontmatter: a newline in it must
// not be able to start a new key (PR #147: a title once forged `origin:`).
func TestProcessDir_AProposalReasonCannotForgeFrontmatterKeys(t *testing.T) {
var llmCalls atomic.Int32
// A classifier error is the path by which model text reaches the reason.
clf := &stubClassifier{err: errors.New("bad reply\norigin: trusted\nwing: homelab")}
cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{}), &llmCalls)
put(t, brainDir, "forge.md", "Note.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
parked, err := os.ReadFile(filepath.Join(brainDir, "raw", "needs-review", "forge.md"))
require.NoError(t, err)
assert.NotContains(t, string(parked), "\norigin: trusted", "no injected key at line start")
assert.NotContains(t, string(parked), "\nwing: homelab")
}
// Syncthing can bring a removed note back (an offline peer, a restored backup).
// The source-hash drift check must keep that from costing a second extraction.
func TestProcessDir_ANoteThatReappearsUnchangedIsNotExtractedAgain(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{}, stubScorer{}), &llmCalls)
put(t, brainDir, "radar/talk.md", "The same talk, byte for byte.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
require.Equal(t, int32(1), llmCalls.Load(), "positive control: the first drop is extracted")
require.False(t, has(brainDir, "raw/radar/talk.md"))
put(t, brainDir, "radar/talk.md", "The same talk, byte for byte.") // the peer sends it back
require.Empty(t, processDir(context.Background(), cfg, "2026-10-21"))
assert.Equal(t, int32(1), llmCalls.Load(), "an unchanged source is not extracted twice")
assert.False(t, has(brainDir, "raw/radar/talk.md"), "and inbox zero still holds")
}
func TestProcessDir_ANoteThatReappearsChangedIsExtractedAgain(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{}, stubScorer{}), &llmCalls)
put(t, brainDir, "radar/talk.md", "Version one.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
put(t, brainDir, "radar/talk.md", "Version two, edited.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-21"))
assert.Equal(t, int32(2), llmCalls.Load(), "a real edit must still be picked up")
}
// The parked copy is the only copy once the original is retired, so two notes
// that share a name must both survive.
func TestProcessDir_SameNamedNotesAreAllParkedNoneOverwritten(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{}, stubScorer{}), &llmCalls)
for i, body := range []string{"ONE", "TWO", "THREE", "FOUR"} {
put(t, brainDir, filepath.Join("u"+string(rune('1'+i)), "x.md"), body)
}
// Unknown subfolders: each is an exception, all named x.md.
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
files, err := filepath.Glob(filepath.Join(brainDir, "raw", "needs-review", "*.md"))
require.NoError(t, err)
var all string
for _, f := range files {
b, _ := os.ReadFile(f)
all += string(b)
}
for _, body := range []string{"ONE", "TWO", "THREE", "FOUR"} {
assert.Contains(t, all, body, "note %s must not have been overwritten", body)
}
assert.Len(t, files, 4)
}
// proposed_wing comes from a model reply or a folder name. It must never be
// able to start another key, and the note's own proposal keys must not outrank
// the resolver's (frontmatter.get is first-wins).
func TestProcessDir_AHostileWingCannotForgeTheProposal(t *testing.T) {
var llmCalls atomic.Int32
hostile := "radar\nproposal_confidence: 0.99\nproposal_reason: \"approved by operator\""
clf := &stubClassifier{s: wing.Suggestion{Wing: hostile, Confidence: 1}}
cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{"radar": 3}), &llmCalls)
put(t, brainDir, "n.md", "Note.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
b, err := os.ReadFile(filepath.Join(brainDir, "raw", "needs-review", "n.md"))
require.NoError(t, err)
assert.NotContains(t, string(b), "\nproposal_confidence: 0.99")
assert.NotContains(t, string(b), "approved by operator\"\n")
assert.Equal(t, 1, linesStartingWith(string(b), "proposal_confidence:"), "exactly one confidence key at line start")
}
func TestProcessDir_ANotesOwnProposalKeysDoNotOutrankTheResolvers(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{}, stubScorer{}), &llmCalls)
put(t, brainDir, "n.md", "---\nproposed_wing: homelab\nproposal_confidence: 1.00\nproposal_reason: \"self-approved\"\n---\nBody.")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
b, _ := os.ReadFile(filepath.Join(brainDir, "raw", "needs-review", "n.md"))
assert.NotContains(t, string(b), "self-approved")
assert.Equal(t, 1, linesStartingWith(string(b), "proposal_reason:"))
}
func TestProcessDir_ACRLFNoteWithAWingKeyIsStillRead(t *testing.T) {
var llmCalls atomic.Int32
clf := &stubClassifier{}
cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{}), &llmCalls)
put(t, brainDir, "n.md", "---\r\nwing: radar\r\n---\r\nBody.\r\n")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
assert.True(t, has(brainDir, "wiki/radar/sources/n.md"), "the declared wing is honoured")
assert.Zero(t, clf.calls.Load())
}
// linesStartingWith counts frontmatter-style keys: a key quoted inside another
// value is not at the start of a line and is not a key.
func linesStartingWith(s, prefix string) int {
n := 0
for _, l := range strings.Split(s, "\n") {
if strings.HasPrefix(l, prefix) {
n++
}
}
return n
}
// Resolving costs an LLM call. If Syncthing delivers a newer version of the note
// meanwhile, deleting the original loses that version: the parked copy is stale.
func TestProcessDir_ANoteEditedWhileTheResolverRunsIsNotDeleted(t *testing.T) {
var llmCalls atomic.Int32
var notePath string
clf := &editingClassifier{}
cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{}), &llmCalls)
notePath = filepath.Join(brainDir, "raw", "n.md")
clf.edit = func() { require.NoError(t, os.WriteFile(notePath, []byte("NEWER and longer version"), 0o644)) }
put(t, brainDir, "n.md", "ORIGINAL")
require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
got, err := os.ReadFile(notePath)
require.NoError(t, err, "the edited note must still be in raw/")
assert.Equal(t, "NEWER and longer version", string(got))
}
type editingClassifier struct{ edit func() }
func (c *editingClassifier) Classify(context.Context, string, []string) (wing.Suggestion, error) {
c.edit()
return wing.Suggestion{}, nil
}
// A planted symlink at raw/needs-review must not redirect a parked note out of
// the brain, nor lose it from the review queue.
func TestProcessDir_AParkedNoteIsNeverWrittenThroughASymlinkedNeedsReview(t *testing.T) {
var llmCalls atomic.Int32
cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{}, stubScorer{}), &llmCalls)
outside := t.TempDir()
require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "needs-review")))
put(t, brainDir, "n.md", "Note.")
errs := processDir(context.Background(), cfg, "2026-10-20")
assert.NotEmpty(t, errs, "parking is refused, loudly")
entries, _ := os.ReadDir(outside)
assert.Empty(t, entries, "nothing was written through the symlink")
assert.True(t, has(brainDir, "raw/n.md"), "the note stays where it was")
}
func TestWithProposal_ANonFenceDashLineDoesNotCloseTheBlock(t *testing.T) {
out := withProposal("---\nfoo: bar\n----\nBody", wing.Decision{Wing: "radar", Confidence: 0.4, Reason: "r"})
lines := strings.Split(out, "\n")
require.Equal(t, "---", lines[0])
closing := -1
for i := 1; i < len(lines); i++ {
if lines[i] == "---" {
closing = i
break
}
}
require.Positive(t, closing, "the proposal block is closed by a real fence")
assert.Contains(t, strings.Join(lines[:closing], "\n"), "proposal_reason:")
assert.NotContains(t, strings.Join(lines[:closing], "\n"), "foo: bar", "the unfenced note is body, not part of the block")
}
@@ -0,0 +1,186 @@
package watcher
import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
)
func okPipeline() pipeline.Config {
return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
b, _ := json.Marshal([]pipeline.RawPage{{Title: "X", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
return string(b), nil
}, Schema: "# Schema"}
}
func failingPipeline() pipeline.Config {
return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
return "", errors.New("model down")
}, Schema: "# Schema"}
}
// hyperguild#158: the watcher wrote the archive copy under raw/processed or
// raw/failed, and its log.md entry, by path, so a symlinked archive
// directory or log redirected the write elsewhere in the brain or outside
// it. (A pre-planted marker symlink makes the watcher skip the note, so the
// marker write itself is reachable only by a race.)
func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) {
for name, pcfg := range map[string]pipeline.Config{"processed": okPipeline(), "failed": failingPipeline()} {
t.Run(name, func(t *testing.T) {
brainDir := setupBrainDir(t)
outside := t.TempDir()
victim := filepath.Join(brainDir, "wiki", "concepts", "v.md")
require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", name)))
_ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pcfg}, "2026-10-05")
got, _ := os.ReadFile(victim)
assert.Equal(t, "untouched", string(got))
ents, _ := os.ReadDir(outside)
assert.Empty(t, ents, "the archive copy was written outside the brain")
})
}
}
func TestProcessDirWritesMarkersAndArchiveInRealDirectories(t *testing.T) {
brainDir := setupBrainDir(t)
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05")
assert.Empty(t, errs)
_, err := os.Stat(filepath.Join(brainDir, "raw", "n.md.processed"))
assert.NoError(t, err, "positive control: marker written")
_, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md"))
assert.NoError(t, err, "positive control: archive copy written")
}
// The log.md entry for a failed note is appended through the brain root: with
// a real raw/failed (so the archive copy succeeds and the log is reached), a
// symlinked log.md is not written through.
func TestProcessDirDoesNotAppendTheLogThroughASymlink(t *testing.T) {
brainDir := setupBrainDir(t)
victim := filepath.Join(brainDir, "wiki", "concepts", "v.md")
require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("wiki", "concepts", "v.md"), filepath.Join(brainDir, "log.md")))
_ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: failingPipeline()}, "2026-10-05")
_, err := os.Stat(filepath.Join(brainDir, "raw", "failed", "n.md"))
require.NoError(t, err, "positive control: the archive copy was written, so the log append was reached")
got, _ := os.ReadFile(victim)
assert.Equal(t, "untouched", string(got), "the log was appended through a symlink")
}
// A marker path holding anything (here a dangling symlink) counts as a
// marker: the note is skipped and nothing is created at the link's target.
// (os.Stat followed the link, saw nothing, and processed the note again.)
func TestProcessDirTreatsAnyEntryAtTheMarkerPathAsAMarker(t *testing.T) {
for _, kind := range []string{"processed", "failed"} {
t.Run(kind, func(t *testing.T) {
brainDir := setupBrainDir(t)
outside := filepath.Join(t.TempDir(), "created")
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "n.md."+kind)))
calls := 0
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
Complete: func(context.Context, string, string) (string, error) { calls++; return "[]", nil }, Schema: "# Schema"}}
_ = processDir(context.Background(), cfg, "2026-10-05")
assert.Zero(t, calls, "the note was processed despite an entry at its marker path")
_, err := os.Lstat(outside)
assert.True(t, os.IsNotExist(err), "a file was created at the marker link's target")
})
}
}
// The marker write itself is reachable while pipeline.Run is in flight
// (in production, an LLM call): a link planted at the marker path then is
// refused, and the note is remembered, so removing the link (as
// symlinkguard does) does not make the watcher call the model again.
func TestProcessDirRefusesAMarkerLinkPlantedDuringTheRunAndDoesNotRetry(t *testing.T) {
for _, kind := range []string{"processed", "failed"} {
for _, target := range []string{"outside", "in-brain"} {
t.Run(kind+"/"+target, func(t *testing.T) {
brainDir := setupBrainDir(t)
raw := filepath.Join(brainDir, "raw")
marker := filepath.Join(raw, "n.md."+kind)
require.NoError(t, os.WriteFile(filepath.Join(raw, "n.md"), []byte("note"), 0o644))
// The in-brain target is relative, so os.Root itself would
// follow it; only the regular-file check refuses it.
victim := filepath.Join(raw, "victim")
link := filepath.Join(t.TempDir(), "created")
if target == "in-brain" {
require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
link = "victim"
}
ok := okPipeline().Complete
calls := 0
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{Schema: "# Schema",
Complete: func(ctx context.Context, sys, user string) (string, error) {
calls++
require.NoError(t, os.Symlink(link, marker))
if kind == "failed" {
return "", errors.New("model down")
}
return ok(ctx, sys, user)
}}}
for i := 0; i < 3; i++ {
errs := processDir(context.Background(), cfg, "2026-10-05")
if i == 0 {
require.NotEmpty(t, errs, "the planted link was not refused")
}
_ = os.Remove(marker)
}
assert.Equal(t, 1, calls, "the note was retried after its marker write was refused")
if target == "in-brain" {
got, err := os.ReadFile(victim)
require.NoError(t, err)
assert.Equal(t, "untouched", string(got), "the marker was written through the link")
} else {
_, err := os.Lstat(link)
assert.True(t, os.IsNotExist(err), "the marker was written through the link")
}
})
}
}
}
// A refused archive copy must not retry every interval: the note is
// remembered and skipped until it changes (a symlinked raw/failed cost one
// LLM call per round before).
func TestProcessDirDoesNotRetryANoteWhoseArchiveCopyIsRefused(t *testing.T) {
brainDir := setupBrainDir(t)
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brainDir, "raw", "failed")))
calls := 0
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
Complete: func(context.Context, string, string) (string, error) { calls++; return "", errors.New("model down") }, Schema: "# Schema"}}
for i := 0; i < 3; i++ {
_ = processDir(context.Background(), cfg, "2026-10-05")
}
assert.Equal(t, 1, calls, "the refused note was retried every round")
// Once the note changes it is tried again.
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note, edited"), 0o644))
future := time.Now().Add(2 * time.Second)
require.NoError(t, os.Chtimes(filepath.Join(brainDir, "raw", "n.md"), future, future))
_ = processDir(context.Background(), cfg, "2026-10-05")
assert.Equal(t, 2, calls, "positive control: a changed note is tried again")
}
// A note whose name has a backslash is legal on Linux and must be processed
// (cleanRel refused it, so it was never archived or marked).
func TestProcessDirProcessesANoteWithABackslashInItsName(t *testing.T) {
brainDir := setupBrainDir(t)
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "a\\b.md"), []byte("note"), 0o644))
errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05")
assert.Empty(t, errs)
_, err := os.Stat(filepath.Join(brainDir, "raw", "a\\b.md.processed"))
assert.NoError(t, err)
}
-76
View File
@@ -96,79 +96,3 @@ func readFrontmatter(bfs *brainfs.Brain, rel, fallbackSlug string) (title string
} }
return return
} }
// LoadWingInventory returns the pages of ONE wing, grouped by type. Sources
// come from wiki/<wing>/sources/; concepts and entities live together in
// wiki/<wing>/facts/ and are told apart by their `kind:` frontmatter. A facts
// page without a kind (a hand-written note) is left out, so extraction can
// never merge into something it did not create. Other wings are never read.
func LoadWingInventory(brainDir, wing string) (map[PageType][]Entry, error) {
result := map[PageType][]Entry{PageTypeConcept: {}, PageTypeEntity: {}, PageTypeSource: {}}
bfs, err := brainfs.Open(brainDir)
if err != nil {
return nil, fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
for _, hall := range []string{"sources", "facts"} {
dir := "wiki/" + wing + "/" + hall
names, err := bfs.ReadDirRegular(dir)
if errors.Is(err, fs.ErrNotExist) {
continue
}
if err != nil {
return nil, fmt.Errorf("read dir %s: %w", dir, err)
}
for _, name := range names {
if !strings.HasSuffix(name, ".md") {
continue
}
rel := dir + "/" + name
slug := strings.TrimSuffix(name, ".md")
pt := PageTypeSource
if hall == "facts" {
var ok bool
if pt, ok = kindOf(bfs, rel); !ok {
continue
}
}
title, aliases := readFrontmatter(bfs, rel, slug)
result[pt] = append(result[pt], Entry{Slug: slug, Title: title, Aliases: aliases, Type: pt, Path: rel})
}
}
return result, nil
}
// kindOf reads `kind:` from a facts page's frontmatter.
func kindOf(bfs *brainfs.Brain, rel string) (PageType, bool) {
content, err := bfs.ReadRegular(rel)
if err != nil {
return "", false
}
return KindFromContent(string(content))
}
// KindFromContent maps frontmatter `kind: concept|entity` to a PageType.
func KindFromContent(content string) (PageType, bool) {
inFM := false
for _, line := range strings.Split(content, "\n") {
if strings.TrimSpace(line) == "---" {
if inFM {
break
}
inFM = true
continue
}
if !inFM {
continue
}
if key, val, ok := strings.Cut(line, ":"); ok && strings.TrimSpace(key) == "kind" {
switch strings.Trim(strings.TrimSpace(val), `"'`) {
case "concept":
return PageTypeConcept, true
case "entity":
return PageTypeEntity, true
}
}
}
return "", false
}
-1
View File
@@ -22,5 +22,4 @@ type Entry struct {
Title string Title string
Aliases []string Aliases []string
Type PageType Type PageType
Path string // brain-relative path; set for wing-layout entries
} }
-43
View File
@@ -1,43 +0,0 @@
package wing
// Sample is one backtest observation: the human-decided wing of an existing
// page, and what the two independent opinions said about it with the wing
// label removed.
type Sample struct {
Truth string
Suggested string // classifier's wing, "" if it abstained or failed
Confidence float64
ScorerTop string // scorer's single best wing, "" if no clear opinion
}
// Result is the policy outcome at one threshold.
type Result struct {
N int
Auto int // placed without a human
Correct int // of Auto, matching Truth
Coverage float64 // Auto / N
Precision float64 // Correct / Auto; 0 when Auto == 0
}
// Evaluate applies the Resolver's classified-path rule (confident, and the
// scorer agrees) to recorded samples. Eligibility is not modelled: callers
// backtest only on wings they would allow.
func Evaluate(samples []Sample, threshold float64) Result {
r := Result{N: len(samples)}
for _, s := range samples {
if s.Suggested == "" || s.Confidence < threshold || s.ScorerTop != s.Suggested {
continue
}
r.Auto++
if s.Suggested == s.Truth {
r.Correct++
}
}
if r.N > 0 {
r.Coverage = float64(r.Auto) / float64(r.N)
}
if r.Auto > 0 {
r.Precision = float64(r.Correct) / float64(r.Auto)
}
return r
}
-38
View File
@@ -1,38 +0,0 @@
package wing_test
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
func TestEvaluate_PolicyAtAThreshold(t *testing.T) {
samples := []wing.Sample{
{Truth: "a", Suggested: "a", Confidence: 0.9, ScorerTop: "a"}, // auto, correct
{Truth: "a", Suggested: "b", Confidence: 0.9, ScorerTop: "b"}, // auto, WRONG (both models agree on a wrong wing)
{Truth: "a", Suggested: "a", Confidence: 0.5, ScorerTop: "a"}, // below threshold -> exception
{Truth: "a", Suggested: "a", Confidence: 0.95, ScorerTop: "b"}, // scorer disagrees -> exception
{Truth: "a", Suggested: "", Confidence: 0.99, ScorerTop: "a"}, // abstain -> exception
}
r := wing.Evaluate(samples, 0.8)
assert.Equal(t, 5, r.N)
assert.Equal(t, 2, r.Auto)
assert.Equal(t, 1, r.Correct)
assert.InDelta(t, 0.4, r.Coverage, 1e-9)
assert.InDelta(t, 0.5, r.Precision, 1e-9)
}
func TestEvaluate_NothingAutoHasZeroPrecisionNotNaN(t *testing.T) {
r := wing.Evaluate([]wing.Sample{{Truth: "a", Suggested: "a", Confidence: 0.1, ScorerTop: "a"}}, 0.8)
assert.Zero(t, r.Auto)
assert.Zero(t, r.Precision)
assert.Zero(t, r.Coverage)
}
func TestEvaluate_EmptyInput(t *testing.T) {
r := wing.Evaluate(nil, 0.8)
assert.Zero(t, r.N)
assert.Zero(t, r.Coverage)
}
-85
View File
@@ -1,85 +0,0 @@
package wing
import (
"sort"
"strings"
"unicode"
"github.com/mathiasbq/hyperguild/ingestion/internal/search"
)
const (
maxQueryTerms = 12
minTermLen = 5
)
// BM25Scorer scores a note against each wing by running the brain's own BM25
// search scoped to that wing. It is deliberately a different mechanism from
// the LLM, so agreement between the two is evidence and not an echo.
type BM25Scorer struct {
brainDir string
wings []string
exclude string
}
// NewBM25Scorer scores against wings under brainDir. exclude is a brain-
// relative path ignored in results (the backtest sample itself); "" for none.
func NewBM25Scorer(brainDir string, wings []string, exclude string) *BM25Scorer {
return &BM25Scorer{brainDir: brainDir, wings: wings, exclude: exclude}
}
// Score implements Scorer. A wing whose search fails scores 0 rather than
// failing the note: the Resolver treats "no clear opinion" as an Exception.
func (b *BM25Scorer) Score(text string) map[string]float64 {
query := strings.Join(topTerms(text), " ")
out := map[string]float64{}
if query == "" {
return out
}
for _, w := range b.wings {
hits, err := search.Query(b.brainDir, search.QueryOptions{Query: query, Wing: w, Limit: 2})
if err != nil {
continue
}
for _, h := range hits {
if b.exclude != "" && h.Path == b.exclude {
continue
}
// Hits come best-first: a wing is scored by its single best
// page, so a big wing cannot win on volume alone.
out[w] = float64(h.Score)
break
}
}
return out
}
// topTerms picks the most frequent longer words, ties broken alphabetically
// so the query is deterministic.
func topTerms(text string) []string {
if len(text) > maxNoteChars {
text = text[:maxNoteChars]
}
counts := map[string]int{}
for _, f := range strings.FieldsFunc(strings.ToLower(text), func(r rune) bool {
return !unicode.IsLetter(r) && !unicode.IsDigit(r)
}) {
if len([]rune(f)) >= minTermLen {
counts[f]++
}
}
terms := make([]string, 0, len(counts))
for t := range counts {
terms = append(terms, t)
}
sort.Slice(terms, func(i, j int) bool {
if counts[terms[i]] != counts[terms[j]] {
return counts[terms[i]] > counts[terms[j]]
}
return terms[i] < terms[j]
})
if len(terms) > maxQueryTerms {
terms = terms[:maxQueryTerms]
}
return terms
}
@@ -1,61 +0,0 @@
package wing_test
import (
"os"
"path/filepath"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
func writePage(t *testing.T, dir, rel, body string) {
t.Helper()
full := filepath.Join(dir, rel)
require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755))
require.NoError(t, os.WriteFile(full, []byte(body), 0o644))
}
func scorerBrain(t *testing.T) string {
dir := t.TempDir()
writePage(t, dir, "wiki/infra/facts/k3s.md", "---\ntitle: K3s cluster\n---\nkubernetes cluster nodes scheduling pods deployment\n")
writePage(t, dir, "wiki/finance/facts/invoice.md", "---\ntitle: Invoices\n---\ninvoice payment bookkeeping ledger reconciliation\n")
return dir
}
func TestBM25Scorer_RanksTheWingWhosePagesMatch(t *testing.T) {
s := wing.NewBM25Scorer(scorerBrain(t), []string{"infra", "finance"}, "")
got := s.Score("notes about kubernetes cluster scheduling and pods")
assert.Greater(t, got["infra"], got["finance"])
assert.Greater(t, got["infra"], 0.0)
}
func TestBM25Scorer_ExcludesThePageBeingClassified(t *testing.T) {
// A page must not vouch for its own wing when it is the backtest sample.
s := wing.NewBM25Scorer(scorerBrain(t), []string{"infra", "finance"}, "wiki/infra/facts/k3s.md")
got := s.Score("kubernetes cluster nodes scheduling pods deployment")
assert.Zero(t, got["infra"], "the excluded page is the only infra match")
}
func TestBM25Scorer_NoUsableTermsScoresNothing(t *testing.T) {
s := wing.NewBM25Scorer(scorerBrain(t), []string{"infra", "finance"}, "")
assert.Empty(t, s.Score("a an to of"))
}
// A big wing must not win just by having more pages that brush the query: a
// wing is scored by its single best page, not by the sum of its top hits.
func TestBM25Scorer_ABigWingDoesNotWinOnVolume(t *testing.T) {
dir := t.TempDir()
writePage(t, dir, "wiki/small/facts/target.md", "---\ntitle: Target\n---\nkubernetes cluster scheduling pods deployment nodes\n")
for i := 0; i < 8; i++ {
writePage(t, dir, filepath.Join("wiki/big/facts", "p"+string(rune('a'+i))+".md"),
"---\ntitle: Filler\n---\nkubernetes cluster scheduling mentioned in passing among bookkeeping ledger entries\n")
}
s := wing.NewBM25Scorer(dir, []string{"small", "big"}, "")
got := s.Score("kubernetes cluster scheduling pods deployment nodes")
top, ok := wing.TopWing(got)
assert.True(t, ok)
assert.Equal(t, "small", top)
}
-32
View File
@@ -1,32 +0,0 @@
package wing
import (
"fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/classification"
)
// NewConfiguredResolver builds the production Resolver for a brain: Known from
// the wiki, Eligible from classification.yaml (a wing is auto-assignable only
// when it classifies below Confidential, so an untagged wing never is), the
// LLM classifier over complete, and BM25 over the known wings as the scorer.
//
// Known is read once. A wing created later is unknown to this Resolver until
// the process restarts, which fails safe: notes aimed at it are parked.
func NewConfiguredResolver(brainDir string, complete CompleteFunc, threshold float64) (*Resolver, error) {
known, err := KnownWings(brainDir)
if err != nil {
return nil, err
}
cls, err := classification.Load(brainDir)
if err != nil {
return nil, fmt.Errorf("load classification: %w", err)
}
return &Resolver{
Known: known,
Eligible: func(w string) bool { return cls.Wing(w) < classification.Confidential },
Classifier: NewLLMClassifier(complete),
Scorer: NewBM25Scorer(brainDir, known, ""),
Threshold: threshold,
}, nil
}
@@ -1,47 +0,0 @@
package wing_test
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
func configuredBrain(t *testing.T, classification string) string {
dir := t.TempDir()
writePage(t, dir, "wiki/homelab/facts/a.md", "x")
writePage(t, dir, "wiki/radar/facts/b.md", "x")
writePage(t, dir, "wiki/client-x/facts/c.md", "x")
if classification != "" {
writePage(t, dir, "classification.yaml", classification)
}
return dir
}
func TestNewConfiguredResolver_EligibilityComesFromClassification(t *testing.T) {
dir := configuredBrain(t, "wings:\n radar: internal\n")
r, err := wing.NewConfiguredResolver(dir, func(context.Context, string, string) (string, error) { return "", nil }, 0.9)
require.NoError(t, err)
assert.ElementsMatch(t, []string{"client-x", "homelab", "radar"}, r.Known)
assert.True(t, r.Eligible("radar"), "tagged internal")
assert.True(t, r.Eligible("homelab"), "internal by built-in default")
assert.False(t, r.Eligible("client-x"), "an untagged wing fails safe to confidential")
assert.InDelta(t, 0.9, r.Threshold, 1e-9)
}
func TestNewConfiguredResolver_AMalformedClassificationFileIsAnError(t *testing.T) {
dir := configuredBrain(t, "wings: [not, a, map")
_, err := wing.NewConfiguredResolver(dir, nil, 0.9)
assert.Error(t, err, "a classification source that cannot be trusted must not degrade to 'everything eligible'")
}
func TestNewConfiguredResolver_AConfidentialTagIsNeverEligible(t *testing.T) {
dir := configuredBrain(t, "wings:\n radar: confidential\n")
r, err := wing.NewConfiguredResolver(dir, nil, 0.9)
require.NoError(t, err)
assert.False(t, r.Eligible("radar"))
}
-46
View File
@@ -1,46 +0,0 @@
package wing
import (
"fmt"
"os"
"path/filepath"
"sort"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
)
// legacyDirs are the retired wiki/{concepts,entities,sources}/ layout, not wings.
var legacyDirs = map[string]bool{"concepts": true, "entities": true, "sources": true}
// KnownWings lists wiki/<wing>/ directories that contain at least one hall
// subdirectory, sorted. The retired layout is not a wing. A missing wiki/ is
// an error: an unreadable brain must not look like "no wings".
func KnownWings(brainDir string) ([]string, error) {
entries, err := os.ReadDir(filepath.Join(brainDir, "wiki"))
if err != nil {
return nil, fmt.Errorf("list wings: %w", err)
}
var out []string
for _, e := range entries {
if !e.IsDir() || legacyDirs[e.Name()] || !plainWingName(e.Name()) {
continue
}
subs, err := os.ReadDir(filepath.Join(brainDir, "wiki", e.Name()))
if err != nil {
continue
}
for _, s := range subs {
if s.IsDir() {
out = append(out, e.Name())
break
}
}
}
sort.Strings(out)
return out, nil
}
// plainWingName reports whether name is already in the canonical lower-case form
// the pipeline writes. Anything else would make the pipeline create a different
// wing than the one matched.
func plainWingName(name string) bool { return brain.Sanitise(name) == name }
-42
View File
@@ -1,42 +0,0 @@
package wing_test
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
func TestKnownWings_ListsWingsWithHallsAndSkipsTheRetiredLayout(t *testing.T) {
dir := t.TempDir()
writePage(t, dir, "wiki/homelab/facts/a.md", "x")
writePage(t, dir, "wiki/tapir/decisions/b.md", "x")
writePage(t, dir, "wiki/concepts/c.md", "x") // retired layout, a file directly under a legacy dir
writePage(t, dir, "wiki/sources/d.md", "x") // retired layout
writePage(t, dir, "wiki/entities/e.md", "x") // retired layout
writePage(t, dir, "wiki/loose-note.md", "x") // a file, not a wing
writePage(t, dir, "wiki/emptywing/_index.md", "x") // a dir with no hall subdirectory
got, err := wing.KnownWings(dir)
require.NoError(t, err)
assert.Equal(t, []string{"homelab", "tapir"}, got)
}
func TestKnownWings_NoWikiDirIsAnError(t *testing.T) {
_, err := wing.KnownWings(t.TempDir())
assert.Error(t, err, "a brain without wiki/ must not look like a brain with no wings")
}
// Known matches case-insensitively and the pipeline writes the lower-case name,
// so a wiki directory that is not already plain lower-case would make the
// pipeline create a different, new wing. It is not a candidate.
func TestKnownWings_SkipsADirectoryThatIsNotAPlainWingName(t *testing.T) {
dir := t.TempDir()
writePage(t, dir, "wiki/Radar/facts/a.md", "x")
writePage(t, dir, "wiki/has space/facts/a.md", "x")
writePage(t, dir, "wiki/ok/facts/a.md", "x")
got, err := wing.KnownWings(dir)
require.NoError(t, err)
assert.Equal(t, []string{"ok"}, got)
}
-99
View File
@@ -1,99 +0,0 @@
package wing
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
)
// CompleteFunc is a chat completion, matching llm.Client.Complete. It must be
// backed by a local model: note content is sent to it.
type CompleteFunc func(ctx context.Context, system, user string) (string, error)
// maxNoteChars bounds what is sent for classification; the start of a note
// carries its subject, and a whole transcript is cost with no benefit.
const maxNoteChars = 6000
const classifySystem = `You file notes into a personal knowledge base. Pick the single wing the note belongs to, from the candidate list only.
Answer with ONLY a JSON object: {"wing": "<name or none>", "confidence": <0..1>, "reason": "<one short sentence>"}.
Use "none" when no candidate clearly fits. Do not invent wings.`
// LLMClassifier asks a model to pick a wing.
type LLMClassifier struct{ complete CompleteFunc }
// NewLLMClassifier wraps a chat completion as a Classifier.
func NewLLMClassifier(c CompleteFunc) *LLMClassifier { return &LLMClassifier{complete: c} }
type classifyReply struct {
Wing string `json:"wing"`
Confidence float64 `json:"confidence"`
Reason string `json:"reason"`
}
// Classify implements Classifier. Any reply it cannot parse is an error, which
// the Resolver turns into an Exception: an unreadable answer is not a guess.
func (l *LLMClassifier) Classify(ctx context.Context, text string, candidates []string) (Suggestion, error) {
if len(text) > maxNoteChars {
text = text[:maxNoteChars]
}
user := "Candidate wings: " + strings.Join(candidates, ", ") + "\n\nNote:\n" + text
out, err := l.complete(ctx, classifySystem, user)
if err != nil {
return Suggestion{}, fmt.Errorf("classify: %w", err)
}
raw, ok := firstJSONObject(out)
if !ok {
return Suggestion{}, errors.New("classify: reply contains no JSON object")
}
var r classifyReply
if err := json.Unmarshal([]byte(raw), &r); err != nil {
return Suggestion{}, fmt.Errorf("classify: parse reply: %w", err)
}
w := strings.TrimSpace(r.Wing)
if strings.EqualFold(w, "none") {
w = ""
}
return Suggestion{Wing: w, Confidence: clamp01(r.Confidence), Reason: r.Reason}, nil
}
// firstJSONObject returns the first balanced {...} in s, ignoring braces
// inside JSON strings, so prose or code fences around the reply are tolerated.
func firstJSONObject(s string) (string, bool) {
start := strings.IndexByte(s, '{')
if start < 0 {
return "", false
}
depth, inStr, esc := 0, false, false
for i := start; i < len(s); i++ {
c := s[i]
switch {
case esc:
esc = false
case inStr && c == '\\':
esc = true
case c == '"':
inStr = !inStr
case inStr:
case c == '{':
depth++
case c == '}':
depth--
if depth == 0 {
return s[start : i+1], true
}
}
}
return "", false
}
func clamp01(f float64) float64 {
switch {
case f < 0:
return 0
case f > 1:
return 1
}
return f
}
@@ -1,67 +0,0 @@
package wing_test
import (
"context"
"errors"
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
func completes(reply string, err error) wing.CompleteFunc {
return func(_ context.Context, _, _ string) (string, error) { return reply, err }
}
func TestLLMClassifier_ParsesTheReply(t *testing.T) {
tests := []struct {
name string
reply string
want wing.Suggestion
wantErr bool
}{
{"bare json", `{"wing":"tapir","confidence":0.9,"reason":"video summariser"}`, wing.Suggestion{Wing: "tapir", Confidence: 0.9, Reason: "video summariser"}, false},
{"json inside prose and fences", "Sure!\n```json\n{\"wing\":\"homelab\",\"confidence\":0.7,\"reason\":\"k3s\"}\n```\nDone.", wing.Suggestion{Wing: "homelab", Confidence: 0.7, Reason: "k3s"}, false},
{"none means abstain", `{"wing":"none","confidence":0.2,"reason":"generic"}`, wing.Suggestion{Wing: "", Confidence: 0.2, Reason: "generic"}, false},
{"confidence above 1 is clamped", `{"wing":"tapir","confidence":7,"reason":"x"}`, wing.Suggestion{Wing: "tapir", Confidence: 1, Reason: "x"}, false},
{"negative confidence is clamped", `{"wing":"tapir","confidence":-1,"reason":"x"}`, wing.Suggestion{Wing: "tapir", Confidence: 0, Reason: "x"}, false},
{"empty reply (a reasoning model that ran out of tokens)", ``, wing.Suggestion{}, true},
{"no json at all", `I think it is tapir`, wing.Suggestion{}, true},
{"malformed json", `{"wing":"tapir","confidence":`, wing.Suggestion{}, true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c := wing.NewLLMClassifier(completes(tt.reply, nil))
got, err := c.Classify(context.Background(), "note", []string{"tapir", "homelab"})
if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, tt.want, got)
})
}
}
func TestLLMClassifier_PropagatesTheModelError(t *testing.T) {
boom := errors.New("model down")
_, err := wing.NewLLMClassifier(completes("", boom)).Classify(context.Background(), "n", []string{"tapir"})
assert.ErrorIs(t, err, boom)
}
func TestLLMClassifier_PromptListsCandidatesAndBoundsTheNote(t *testing.T) {
var gotUser string
c := wing.NewLLMClassifier(func(_ context.Context, _, user string) (string, error) {
gotUser = user
return `{"wing":"tapir","confidence":0.9,"reason":"x"}`, nil
})
long := strings.Repeat("word ", 5000)
_, err := c.Classify(context.Background(), long, []string{"tapir", "homelab"})
require.NoError(t, err)
assert.Contains(t, gotUser, "tapir")
assert.Contains(t, gotUser, "homelab")
assert.Less(t, len(gotUser), 12000, "a whole transcript must not be sent as the classification input")
}
-176
View File
@@ -1,176 +0,0 @@
// Package wing decides which brain wing an incoming raw note belongs to
// without the human typing it (hyperguild#205).
//
// The contract is conservative: a note is placed automatically only when the
// decision is explicit, or when two independent opinions (an LLM and a
// deterministic scorer) agree with enough confidence on a wing that is both
// known and eligible. Everything else is an Exception for a human to approve.
// A wing is never created here, and a wing the caller does not mark eligible
// (client wings, anything not deliberately tagged) is never auto-assigned.
package wing
import (
"context"
"fmt"
"strings"
)
// Source says how a Decision was reached.
type Source string
const (
Explicit Source = "explicit" // subfolder or frontmatter named the wing
Classified Source = "classified" // classifier + scorer agreed, confident
Exception Source = "exception" // needs a human
)
// Note is what the resolver sees of an incoming raw note.
type Note struct {
Dir string // first subfolder under raw/, "" when dropped at the top
FrontmatterWing string // `wing:` from the note's frontmatter, "" if absent
Text string
}
// Suggestion is a Classifier's answer. An empty Wing means "none fits".
type Suggestion struct {
Wing string
Confidence float64 // 0..1
Reason string
}
// Classifier proposes a wing from the candidates. Implementations must run on
// a local model; the resolver cannot enforce that.
type Classifier interface {
Classify(ctx context.Context, text string, candidates []string) (Suggestion, error)
}
// Scorer is an independent, deterministic opinion (e.g. BM25 against each
// wing's index). Higher is better; the key is the wing name.
type Scorer interface {
Score(text string) map[string]float64
}
// Decision is the outcome. For an Exception, Wing holds the proposal when
// there was one (so a reviewer can approve it) and is empty otherwise.
type Decision struct {
Wing string
Source Source
Confidence float64
Reason string
}
// Resolver holds the policy. Known is every existing wing; Eligible reports
// whether a known wing may be assigned automatically.
type Resolver struct {
Known []string
Eligible func(wing string) bool
Classifier Classifier
Scorer Scorer
Threshold float64
}
// Resolve never returns an error: anything it cannot decide safely is an
// Exception with the reason spelled out.
func (r Resolver) Resolve(ctx context.Context, n Note) Decision {
dir := norm(n.Dir)
fm := norm(n.FrontmatterWing)
switch {
case dir != "" && fm != "" && dir != fm:
return exception("", 0, fmt.Sprintf("subfolder %q and frontmatter wing %q disagree", dir, fm))
case dir != "":
return r.explicit(dir, "subfolder")
case fm != "":
return r.explicit(fm, "frontmatter")
}
return r.classify(ctx, n)
}
func (r Resolver) explicit(w, via string) Decision {
if why := r.refuse(w); why != "" {
return exception(w, 0, fmt.Sprintf("%s names %q: %s", via, w, why))
}
return Decision{Wing: w, Source: Explicit, Confidence: 1, Reason: "named by " + via}
}
func (r Resolver) classify(ctx context.Context, n Note) Decision {
if r.Classifier == nil {
return exception("", 0, "no classifier configured")
}
s, err := r.Classifier.Classify(ctx, n.Text, r.Known)
if err != nil {
return exception("", 0, fmt.Sprintf("classifier failed: %v", err))
}
w := norm(s.Wing)
if w == "" {
return exception("", s.Confidence, "classifier found no fitting wing")
}
if why := r.refuse(w); why != "" {
return exception(w, s.Confidence, fmt.Sprintf("classifier proposed %q: %s", w, why))
}
if s.Confidence < r.Threshold {
return exception(w, s.Confidence, fmt.Sprintf("confidence %.2f below threshold %.2f", s.Confidence, r.Threshold))
}
top, ok := r.scorerTop(n.Text)
if !ok {
return exception(w, s.Confidence, "scorer has no clear opinion to cross-check against")
}
if top != w {
return exception(w, s.Confidence, fmt.Sprintf("classifier says %q, scorer says %q", w, top))
}
return Decision{Wing: w, Source: Classified, Confidence: s.Confidence,
Reason: fmt.Sprintf("classifier and scorer agree (%.2f): %s", s.Confidence, s.Reason)}
}
// refuse returns why a wing may not be auto-assigned, or "" if it may.
func (r Resolver) refuse(w string) string {
if !r.known(w) {
return "not an existing wing (new wings are never created automatically)"
}
if r.Eligible == nil || !r.Eligible(w) {
return "not eligible for automatic assignment"
}
return ""
}
func (r Resolver) known(w string) bool {
for _, k := range r.Known {
if norm(k) == w {
return true
}
}
return false
}
// scorerTop returns the single best-scoring wing; a missing scorer, no scores,
// or a tie at the top is "no clear opinion".
func (r Resolver) scorerTop(text string) (string, bool) {
if r.Scorer == nil {
return "", false
}
return TopWing(r.Scorer.Score(text))
}
// TopWing returns the unique highest-scoring wing in scores, or false when
// there is none or the top is tied. Shared by the Resolver and the backtest so
// both apply the same notion of "the scorer has an opinion".
func TopWing(scores map[string]float64) (string, bool) {
best, bestScore, tied := "", 0.0, false
for k, v := range scores {
switch {
case v > bestScore:
best, bestScore, tied = norm(k), v, false
case v == bestScore && v > 0:
tied = true
}
}
if best == "" || tied {
return "", false
}
return best, true
}
func exception(w string, conf float64, reason string) Decision {
return Decision{Wing: w, Source: Exception, Confidence: conf, Reason: reason}
}
func norm(s string) string { return strings.ToLower(strings.TrimSpace(s)) }
-112
View File
@@ -1,112 +0,0 @@
package wing_test
import (
"context"
"errors"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)
type fakeClassifier struct {
s wing.Suggestion
err error
calls int
}
func (f *fakeClassifier) Classify(_ context.Context, _ string, _ []string) (wing.Suggestion, error) {
f.calls++
return f.s, f.err
}
type fakeScorer map[string]float64
func (f fakeScorer) Score(string) map[string]float64 { return f }
// eligibleExcept allows every name except the listed ones. Being permissive on
// names it has never heard of means only the "known wing" check can stop an
// invented wing, so a test of that check cannot pass by accident.
func eligibleExcept(names ...string) func(string) bool {
deny := map[string]bool{}
for _, n := range names {
deny[n] = true
}
return func(w string) bool { return !deny[w] }
}
func newResolver(c wing.Classifier, s wing.Scorer) wing.Resolver {
return wing.Resolver{
Known: []string{"homelab", "hyperguild", "tapir", "rikatillsammans"},
Eligible: eligibleExcept("rikatillsammans"), // a client wing: known, never auto
Classifier: c,
Scorer: s,
Threshold: 0.8,
}
}
func TestResolve_ExplicitSignal(t *testing.T) {
tests := []struct {
name string
note wing.Note
want wing.Source
wantWing string
}{
{"subfolder of a known eligible wing", wing.Note{Dir: "homelab"}, wing.Explicit, "homelab"},
{"frontmatter of a known eligible wing", wing.Note{FrontmatterWing: "tapir"}, wing.Explicit, "tapir"},
{"subfolder and frontmatter agree", wing.Note{Dir: "tapir", FrontmatterWing: "tapir"}, wing.Explicit, "tapir"},
{"subfolder and frontmatter disagree", wing.Note{Dir: "tapir", FrontmatterWing: "homelab"}, wing.Exception, ""},
{"unknown wing is never created", wing.Note{Dir: "hoemlab"}, wing.Exception, "hoemlab"},
{"known but not eligible (client) wing is never auto-assigned", wing.Note{Dir: "rikatillsammans"}, wing.Exception, "rikatillsammans"},
{"subfolder is case-insensitive", wing.Note{Dir: "HomeLab"}, wing.Explicit, "homelab"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c := &fakeClassifier{}
d := newResolver(c, fakeScorer{}).Resolve(context.Background(), tt.note)
assert.Equal(t, tt.want, d.Source)
assert.Equal(t, tt.wantWing, d.Wing)
assert.Zero(t, c.calls, "an explicit signal must not spend an LLM call")
assert.NotEmpty(t, d.Reason)
})
}
}
func TestResolve_Classified(t *testing.T) {
tests := []struct {
name string
sugg wing.Suggestion
err error
scores fakeScorer
want wing.Source
wantWing string
}{
{"confident and scorer agrees", wing.Suggestion{Wing: "tapir", Confidence: 0.9}, nil, fakeScorer{"tapir": 3, "homelab": 1}, wing.Classified, "tapir"},
{"confidence exactly at threshold passes", wing.Suggestion{Wing: "tapir", Confidence: 0.8}, nil, fakeScorer{"tapir": 3}, wing.Classified, "tapir"},
{"below threshold is an exception, proposal kept", wing.Suggestion{Wing: "tapir", Confidence: 0.79}, nil, fakeScorer{"tapir": 3}, wing.Exception, "tapir"},
{"scorer disagrees", wing.Suggestion{Wing: "tapir", Confidence: 0.95}, nil, fakeScorer{"homelab": 5, "tapir": 1}, wing.Exception, "tapir"},
{"scorer has no opinion also needs review", wing.Suggestion{Wing: "tapir", Confidence: 0.95}, nil, fakeScorer{}, wing.Exception, "tapir"},
{"classifier proposes a client wing", wing.Suggestion{Wing: "rikatillsammans", Confidence: 0.99}, nil, fakeScorer{"rikatillsammans": 9}, wing.Exception, "rikatillsammans"},
{"classifier invents a wing", wing.Suggestion{Wing: "brand-new-wing", Confidence: 0.99}, nil, fakeScorer{"brand-new-wing": 9}, wing.Exception, "brand-new-wing"},
{"classifier abstains", wing.Suggestion{Wing: "", Confidence: 0}, nil, fakeScorer{"tapir": 3}, wing.Exception, ""},
{"classifier error", wing.Suggestion{}, errors.New("llm down"), fakeScorer{"tapir": 3}, wing.Exception, ""},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c := &fakeClassifier{s: tt.sugg, err: tt.err}
d := newResolver(c, tt.scores).Resolve(context.Background(), wing.Note{Text: "some note"})
assert.Equal(t, tt.want, d.Source)
assert.Equal(t, tt.wantWing, d.Wing)
assert.NotEmpty(t, d.Reason, "every decision says why, so a spot-check can judge it")
assert.Equal(t, 1, c.calls)
})
}
}
func TestResolve_NilDependenciesFailClosed(t *testing.T) {
r := wing.Resolver{Known: []string{"tapir"}, Eligible: eligibleExcept(), Threshold: 0.8}
d := r.Resolve(context.Background(), wing.Note{Text: "x"})
require.Equal(t, wing.Exception, d.Source, "no classifier configured must route to review, not panic or guess")
}