lachtan/nanobot-runtime
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update projektu

Browse Source
This commit is contained in:
lachtan
2026-07-22 12:32:02 +02:00
parent 19014ed3d9
commit 8e66d6b92a
22 changed files with 1995 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files

243
skills/note/SKILL.md
View File

@@ -1,151 +1,178 @@
---
name: note
description: >
Explicit notes.
Use when user says "note X", "note it".
Capture notes, texts, URLs, or whole articles into a personal knowledge base and
answer questions against it. Triggers on "note X", "/note cron X", "search my
notes for X", "delete/edit the note about X", "forget X". For filing reference
material to search later — not a short durable fact/preference to just
remember, and not a bare URL saved only to read later with no filing.
---
# Note
Explicit note store backed by SQLite. User says "note X" → take only
explicitly-typed tags, reformulate content, store via `note.py add`. Delete only
on explicit user request. Notes are stored to sqlite db.
A personal capture-to-knowledge-base skill. The user throws in notes, texts, URLs, or
whole articles from any channel; each input is captured raw, then reformulated and filed
into one structured markdown document (`notes/notes.md`) organized into thematic sections
that the LLM owns and grows. Search = load the whole document and answer from it.
## Backend
## Architecture — read this first
`skills/note/scripts/note.py` — CLI wrapper around `db/note.sqlite`.
Operation log: `log/note.log` (append-only, all write operations).
Two-stage pipeline, one shared compile step:
## Tag protocol
- **Capture (always, instant, dumb).** `note_capture.py` writes the raw input verbatim
into `notes/inbox/` (atomic) plus one line to `log/note.log`. No reformulation, no
reading of `notes.md`, no fetching. This is all capture ever does.
- **Compile (reformulate + file into `notes/notes.md`).** Runs either **inline** in the
immediate mode, or in the **background** cron drain. Same workflow either way.
Tags are the **first token** right after the trigger — comma-separated, no spaces:
Storage layout under the workspace root (fixed locations):
```
/note arch explanation of the architecture decision → tags: [arch]
/note hw,linux interesting article about kernel → tags: [hw, linux]
/note this is a note without tags → tags: []
```text
notes/
├── notes.md ← THE structured doc (thematic ## sections, LLM-owned)
├── inbox/ ← pending captures (one file each); compile drains this
├── done/ ← successfully compiled captures (sibling of inbox/)
├── hard/ ← held back: paywalled / unreadable / ambiguous — for manual review
└── .compile.lock ← concurrency lock shared by inline compile and cron drain
log/note.log ← append-only audit of every capture
```
Rules:
- **Tags come *only* from the first token the user actually typed. Never
derive, infer, or invent tags from the note's content, topic, or meaning.**
If the user did not type a tag, the note has no tags — full stop.
- Lowercase only; multi-word tags use `-`: `cli`, `soft-delete`, `task-queue`
- If user writes `#tag`, strip `#` before passing to the script
- If no tag is given — that is fine, use no tags; never force tags
**No database — ever.** Notes live *only* in `notes/notes.md` (prose) plus the `notes/`
pipeline dirs above. There is **no** SQLite/DB backend. To find a note, `grep` or read
`notes/notes.md` — **never search `db/`, never run `sqlite3`, never create or open any
`.sqlite`/`.db` file.** (The AGENTS.md "store SQLite under `db/`" convention does **not**
apply to notes — that is for other skills.) An older version of this skill used a
database; it is gone. If you catch yourself opening a DB, stop — the answer is in
`notes/notes.md`.
Tags must be **registered before use**. There is no auto-creation: the database
holds a registry of known tags, and `add` rejects any tag that is not in it (exit
2). A new tag is born only via the explicit `tag-add` command (see Tag management).
Still only pass tags the user typed — registration does not license inventing them.
**Separate store.** `notes/` is not agent memory: keep it distinct from `keep`,
`MEMORY.md`, and the llm-wiki store (`cml/`). Never cross-read or cross-write between
them. The Dream processor must not touch `notes/`.
## Write protocol
**Run scripts with `uv run`, workspace-relative paths** (exec runs from the workspace
root, not the skill dir): `uv run skills/note/scripts/<script>.py …`.
1. Take inline tags from the first token only (see Tag protocol above). If that
token is not a tag the user typed, the tags field stays empty — never fill it
from the content.
2. Reformulate the remaining text into a terse fact. One concept per entry —
split if too complex; omit context that is not itself a fact. Preserve
input language; never translate. Drop filler.
- Input: "poznamenej si, glow zobrazuje markdown v terminálu #cli"
- Run: `uv run skills/note/scripts/note.py add "glow displays markdown in terminal" --tags cli`
3. **Unknown tag (`add` exits 2, prints `Unknown tag(s): …`):** the note was NOT
stored. For each unknown tag, ask the user (in their language): "Tag #X
doesn't exist — create it?"
- **Yes** → `uv run skills/note/scripts/note.py tag-add X`, then re-run `add`
with the original tags.
- **No** → re-run `add` without that tag (keep the known ones). If nothing
remains, store with no tags.
4. Echo: `Noted [#1]: <content> [#tag1 #tag2]` (tags omitted if none).
`#1` is the display ID of the new note — use it to delete immediately if needed.
**Language.** This skill body is English; always reply to the user in the user's own
language.
No dedup. No MEMORY.md lookup. Blind append.
## `/note <text>` — capture and file NOW (default)
## Tag management
The default: file the note into the knowledge base immediately, in this turn.
Tags are created and listed explicitly — never as a side effect of adding a note.
1. Read `Channel` / `Chat ID` from the runtime context if present.
2. Capture:
`uv run skills/note/scripts/note_capture.py --text "<raw input>" [--channel <ch>] [--chat-id <id>]`
Pass the input **as-is** — do not reformulate or strip URLs here.
3. Run the **Compile workflow** (below) inline: acquire the lock, process `notes/inbox/`,
file into `notes/notes.md`, move the source to `notes/done/` (or `notes/hard/`).
4. **Commit** the change (see *Versioning* below): via `exec` run
`git add notes/ && git commit -m "note: <short summary of what landed>"`.
5. Confirm to the user **which section** it landed under, **and quote the exact
text that was filed** (the reformulated fact(s), verbatim as written into
`notes.md` — not a re-summary of it), in their language.
Trigger (create): `/note tag add X`, "create tag X", "register tag X".
This blocks the turn for a while (reads the whole doc; a URL/article adds a fetch). If
the user is firing off many notes quickly, suggest `/note cron`.
1. Run: `uv run skills/note/scripts/note.py tag-add X`
2. Echo the result. Already-existing tag → script reports it and exits 0 (no error).
3. No tag name given → ask which tag to create; do not guess.
## `/note cron <text>` — deferred capture
Trigger (list): `/note tags`, "what tags are there?", "list tags".
Capture only; let the background cron file it later. Fast, non-blocking.
1. Run: `uv run skills/note/scripts/note.py tag-list`
2. Echo output. Empty → "No tags."
1. Read `Channel` / `Chat ID` from the runtime context if present.
2. `uv run skills/note/scripts/note_capture.py --text "<raw input>" [--channel <ch>] [--chat-id <id>]`
3. Confirm in **one short line** (e.g. "captured — I'll file it in the background") and **STOP the
turn**. Forbidden here: reformulating, reading `notes/notes.md`, running any compile
step, taking the lock. If you catch yourself about to read the doc, you are compiling
inline — stop and just capture.
Tags are referenced by name everywhere (no display ID). There is no tag deletion.
## Compile workflow (shared: inline immediate mode + cron drain)
## List protocol
The cron (`note_compile.py`) invokes this via a drain goal; immediate mode runs it inline.
Either way:
Trigger: `/note list`, `show notes`, `what notes do you have?`
1. **Take the lock.** Create `notes/.compile.lock` (skip if a live one exists — another
compile is running; try again later). The cron script handles this itself; inline mode
must respect it so an inline merge and a cron drain never edit `notes.md` at once.
2. **For each file in `notes/inbox/`:**
- **Reformulate** the body into a terse fact (or a few). One concept per entry; drop
filler; **preserve the input language** — never translate. Split if too complex.
If the input is Czech typed without diacritics (e.g. "kdyz uz to psal bez hacku"),
restore correct diacritics as part of reformulation. Leave already-accented text
and non-Czech text untouched — never add diacritics where none belong.
- **URLs:** extract **every** URL from the body (0..N). Fetch each with the `web` tool
(Jina Reader — returns clean markdown, handles JS and soft paywalls). If a URL is
**paywalled / login-gated / truncated / unreadable** (login/subscribe/metered
content, very short output, HTTP 401/403): **do not fabricate a summary** — write
just the URL + any available title + a `⚠ paywall/incomplete` marker. Whole articles:
summarize the key points.
- **File it** under the right thematic `##` section of `notes/notes.md`. Create a new
section if none fits. Use a surgical `str_replace`/append — never rewrite the whole
document.
- **Move the source out of `inbox/` immediately:** to `notes/done/` if anything usable
was filed (paywall markers count as filed — they are the breadcrumb); to `notes/hard/`
if nothing usable could be extracted. Move right after each file so a crash mid-batch
re-processes at most one.
3. **Release the lock** (the cron script does this in `finally`).
1. Run: `uv run skills/note/scripts/note.py list [--limit N] [--tag TAG [TAG ...]]`
2. Echo output. If empty → respond "No notes."
Never assert content you could not read. When unsure, hedge or mark it.
`--tag` accepts one or more tags; OR logic (notes with at least one matching tag).
## `/note search <query>` / `/note find <query>` — query
The number before each note (`1.`, `2.`, …) is the **display ID** — sequential
among active notes, newest first. Renumbers after every deletion. Never change,
renumber, or drop it.
Also triggered by "what do I have on …?", "find in my notes …".
### URLs in a note
1. Read the whole `notes/notes.md` — that single file **is** the knowledge base. Do
not read `inbox/`, `done/`, or `hard/` (those are the raw pipeline, not the KB).
2. Answer from it. If the topic is not covered, say so plainly — do not confabulate.
3. Read-only: never modify the document in a search turn.
The script already lays out each URL (with its inline label, if any) on its own
indented bullet line. **Echo the output verbatim** — keep the bullets and line
breaks, keep URLs bare. Never collapse the bullets back onto one line and never
wrap a URL in `[text](url)`: this chat UI merges two adjacent inline links into
one block, hides the second URL, and overlays the list number. Bare URLs on their
own lines autolink correctly and stay separate.
## `/note delete <query>` / `/note edit <query>` — remove or change a note
## Show protocol
Also triggered by "delete/remove the note about X", "forget X", "edit/update the note
about X". Notes are prose in `notes/notes.md` — **no IDs, no DB.** The user
names a note by describing it; you find it by reading the document.
Trigger: `/note show <id>`, `show note N`, `read note N`, `what does note N say`.
**This is a hard-gated TWO-TURN flow. NEVER delete or edit in the same turn as the
request — showing is not doing.**
1. Display IDs are the same as in `list`/`delete` — sequential among active
notes, newest first, renumbered after every deletion. If unsure, run `list`
first.
2. Run: `uv run skills/note/scripts/note.py show <display-id>`
- Exit 0 → **output the script's stdout verbatim — print every line exactly
as emitted.** Do not summarize, shorten, rewrap, or drop any part of the
`content` field, including URLs and links. The `show` command exists
precisely to surface the note in full; brevity directives do not apply here.
- Exit 1 → display ID out of range; respond accordingly.
3. `show` is read-only — it never deletes or modifies anything.
**Turn 1 — locate and confirm (absolutely NO mutation):**
The block contains every stored field: display ID, internal DB id, creation
timestamp, tags, and full untruncated content.
1. Read `notes/notes.md` to find the matching note(s). That file is the only place a note
lives — **do not search `db/`, do not run `sqlite3`, do not read `inbox/`/`done/`/
`hard/`.** If nothing matches, say so. If the target is ambiguous or several entries
match, list the candidates and ask which one.
2. Show the user the **exact verbatim line(s)/section** you would remove (for an edit: the
`before` → `after`), and ask them to confirm in plain words. Then **STOP the turn.**
Forbidden this turn: `str_replace`/`edit_file`, `rm`, `git`, or any other mutation.
## Delete protocol
**Turn 2 — only after the user explicitly confirms ("yes", "confirmed", "delete it"…):**
Trigger: `/note delete`, `delete a note`, `remove a note`.
1. Remove/change it in `notes/notes.md` with a surgical `str_replace` (never rewrite the
whole document). **Touch `notes/notes.md` only** — do NOT delete or move anything in
`notes/done/`; those raw-capture breadcrumbs are internal plumbing, not a second copy
of the note.
2. Commit (see *Versioning*): via `exec` run
`git add notes/ && git commit -m "note: delete <short desc>"` (edit → `note: edit …`).
1. If the user has not specified an ID, run `list` first to show current notes.
2. Run: `uv run skills/note/scripts/note.py delete <display-id>`
- Exit 0 → confirm deletion.
- Exit 1 → display ID out of range; respond accordingly.
3. Nothing is deleted automatically. Only this explicit protocol deletes.
**Exception:** a capture still **pending** (not yet compiled, a file in `notes/inbox/`)
never reached the KB — you may cancel it directly with `exec: rm notes/inbox/<file>`
(commit only if it was already tracked).
Display IDs renumber after every deletion (e.g., after deleting #3, the old #4
becomes #3). Always run `list` first if unsure of current IDs.
## Versioning (git)
`notes/` lives inside the workspace git repo. The Dream processor never touches it, so
**this skill is the only thing that commits `notes/`** — do it after every change to
`notes/notes.md`:
- **Inline `/note <text>` and delete/edit:** commit in the same turn via `exec`
(`git add notes/ && git commit -m "note: …"`). One commit per operation.
- **Cron drain:** `note_compile.py` commits deterministically after the batch — you do
not commit inside the cron `DRAIN_GOAL` run.
- Never `git add -A` (Dream owns the rest of the workspace); stage only `notes/`. The
`.compile.lock` is gitignored, so `git add notes/` never stages it.
## Edge cases
- `/note` with no content → ask "What should I note?"
- Vague input → ask for the concrete fact; do not store a placeholder.
- `/note tag add` with no name → ask which tag to create; never guess.
- `/note show` with no ID → run `list` first, then ask which display ID.
- `/note delete` with no ID → run `list` first, then ask which display ID.
- Multi-line input → collapse to one line; one entry = one row.
## Rules
- Never store verbatim input. Always reformulate. Preserve input language.
- Do not store smalltalk or meta-commentary about the note skill itself.
- **No auto-load:** `note.sqlite` is never referenced in bootstrap files.
- **No auto-delete / no compaction.** Only explicit delete marks an entry.
- **Delete is soft** — the entry is marked with a timestamp, not removed from
the database. The operation log (`log/note.log`) is the primary audit trail.
- Separate from `/keep`, `MEMORY.md`, Dream — never cross-write or cross-read.
- `/note` with no content → ask what to note.
- Empty / whitespace-only input → `note_capture.py` exits non-zero; ask for real content.
- The compile step, not capture, decides sections and does all fetching. If you ever find
yourself reformulating or reading `notes.md` during a `cron` capture, stop.

0
skills/note/notes.db
View File

316
skills/note/scripts/note.py
View File

@@ -1,316 +0,0 @@
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = []
# ///
"""
note.py — backend for /note skill.
SQLite-backed note store with tags, soft-delete, and operation log.
"""
import argparse
import json
import re
import sqlite3
import sys
from collections.abc import Generator
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
DB_PATH = Path(__file__).resolve().parent.parent.parent.parent / "db" / "note.sqlite"
LOG_PATH = Path(__file__).resolve().parent.parent.parent.parent / "log" / "note.log"
_TAG_RE = re.compile(r"^[a-z][a-z0-9-]*$")
# A URL together with an immediately preceding "Label:" token, if any.
# The leading separator class swallows the connector that introduced the URL
# (em-dash, comma, etc.) so it does not dangle once the URL moves to its own line.
_LABELED_URL_RE = re.compile(r"[\s,;—–-]*([^\s,]+:\s*)?(https?://[^\s,]+)")
SCHEMA = """
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
tags TEXT NOT NULL DEFAULT '[]',
created_at TEXT NOT NULL,
deleted_at TEXT
);
CREATE TABLE IF NOT EXISTS tags (
name TEXT PRIMARY KEY,
created_at TEXT NOT NULL
);
"""
def _init_db(conn: sqlite3.Connection) -> None:
conn.execute("PRAGMA journal_mode=WAL")
conn.executescript(SCHEMA)
_migrate(conn)
def _migrate(conn: sqlite3.Connection) -> None:
cols = {row[1] for row in conn.execute("PRAGMA table_info(notes)")}
if "tags" not in cols:
conn.execute("ALTER TABLE notes ADD COLUMN tags TEXT NOT NULL DEFAULT '[]'")
if "deleted_at" not in cols:
conn.execute("ALTER TABLE notes ADD COLUMN deleted_at TEXT")
conn.commit()
_backfill_tags(conn)
def _backfill_tags(conn: sqlite3.Connection) -> None:
"""On first introduction of the registry, seed it from tags already used in notes."""
existing = {row[0] for row in conn.execute("SELECT name FROM tags")}
if existing:
return
used = {row[0] for row in conn.execute("SELECT DISTINCT value FROM notes, json_each(notes.tags)")}
if not used:
return
now = datetime.now(timezone.utc).isoformat()
conn.executemany(
"INSERT OR IGNORE INTO tags(name, created_at) VALUES(?, ?)",
[(tag, now) for tag in sorted(used)],
)
conn.commit()
@contextmanager
def _connect() -> Generator[sqlite3.Connection, None, None]:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
_init_db(conn)
try:
yield conn
finally:
conn.close()
def _validate_tags(tags: list[str]) -> None:
for tag in tags:
if not _TAG_RE.match(tag):
raise ValueError(
f"Invalid tag '{tag}' — use lowercase letters, digits, hyphens only (e.g. cli, soft-delete)"
)
def _tags_display(tags_json: str) -> str:
tags = json.loads(tags_json)
if not tags:
return ""
return " [" + " ".join(f"#{t}" for t in tags) + "]"
def _urls_on_own_lines(text: str) -> str:
"""Lay out each URL (and its inline "Label:", if any) on its own bullet line.
The chat UI merges two adjacent links into one block and hides the second,
which also overlays the list number. Putting each URL on its own line keeps
them separate and the number visible. URLs stay bare so they autolink.
"""
if not _LABELED_URL_RE.search(text):
return text
def repl(match: re.Match[str]) -> str:
label = match.group(1) or ""
return f"\n - {label}{match.group(2)}"
return _LABELED_URL_RE.sub(repl, text)
def _log(op: str, detail: str) -> None:
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
with LOG_PATH.open("a") as f:
f.write(f"{ts} {op} {detail}\n")
def _active_ids(conn: sqlite3.Connection) -> list[int]:
rows = conn.execute(
"SELECT id FROM notes WHERE deleted_at IS NULL ORDER BY created_at DESC"
).fetchall()
return [row["id"] for row in rows]
def cmd_add(args: argparse.Namespace) -> int:
tags: list[str] = args.tags or []
try:
_validate_tags(tags)
except ValueError as exc:
print(str(exc), file=sys.stderr)
return 1
content = args.text.strip()
tags_json = json.dumps(tags)
created_at = datetime.now(timezone.utc).isoformat()
with _connect() as conn:
known = {row[0] for row in conn.execute("SELECT name FROM tags")}
unknown = [tag for tag in tags if tag not in known]
if unknown:
print(f"Unknown tag(s): {', '.join(unknown)}", file=sys.stderr)
return 2
cur = conn.execute(
"INSERT INTO notes(content, tags, created_at) VALUES(?, ?, ?)",
(content, tags_json, created_at),
)
conn.commit()
nid = cur.lastrowid
tags_log = ",".join(tags)
_log("ADD", f"id={nid} tags=[{tags_log}] {content}")
print(f"Noted [#1]: {content}{_tags_display(tags_json)}")
return 0
def cmd_list(args: argparse.Namespace) -> int:
with _connect() as conn:
id_to_display = {nid: i + 1 for i, nid in enumerate(_active_ids(conn))}
if args.tag:
placeholders = ",".join("?" * len(args.tag))
rows = conn.execute(
f"""
SELECT id, content, tags FROM notes
WHERE deleted_at IS NULL
AND (
SELECT count(*) FROM json_each(notes.tags)
WHERE value IN ({placeholders})
) > 0
ORDER BY created_at DESC
LIMIT ? OFFSET ?
""",
(*args.tag, args.limit, args.offset),
).fetchall()
else:
rows = conn.execute(
"SELECT id, content, tags FROM notes"
" WHERE deleted_at IS NULL"
" ORDER BY created_at DESC LIMIT ? OFFSET ?",
(args.limit, args.offset),
).fetchall()
tag_filter = ",".join(args.tag) if args.tag else "None"
_log("LIST", f"tag={tag_filter} returned={len(rows)}")
if not rows:
print("No notes.")
return 0
for row in rows:
head, sep, rest = _urls_on_own_lines(row["content"]).partition("\n")
print(f"{id_to_display[row['id']]}. {head}{_tags_display(row['tags'])}{sep}{rest}")
return 0
def cmd_delete(args: argparse.Namespace) -> int:
display_id: int = args.id
deleted_at = datetime.now(timezone.utc).isoformat()
with _connect() as conn:
ids = _active_ids(conn)
idx = display_id - 1
if idx < 0 or idx >= len(ids):
print(f"No active note with display id={display_id}.")
return 1
nid = ids[idx]
row = conn.execute(
"SELECT id, content, tags FROM notes WHERE id = ?", (nid,)
).fetchone()
conn.execute("UPDATE notes SET deleted_at = ? WHERE id = ?", (deleted_at, nid))
conn.commit()
tags_log = ",".join(json.loads(row["tags"]))
_log("DELETE", f"display_id={display_id} id={nid} tags=[{tags_log}] content={row['content']!r}")
print(f"Deleted: {row['content']}")
return 0
def cmd_show(args: argparse.Namespace) -> int:
display_id: int = args.id
with _connect() as conn:
ids = _active_ids(conn)
idx = display_id - 1
if idx < 0 or idx >= len(ids):
print(f"No active note with display id={display_id}.")
return 1
nid = ids[idx]
row = conn.execute(
"SELECT id, content, tags, created_at FROM notes WHERE id = ?", (nid,)
).fetchone()
_log("SHOW", f"display_id={display_id} id={nid}")
tags = json.loads(row["tags"])
tags_line = " ".join(f"#{t}" for t in tags) if tags else "(none)"
print(f"Note [#{display_id}] (id={row['id']})")
print(f"created: {row['created_at']}")
print(f"tags: {tags_line}")
print(f"content: {row['content']}")
return 0
def cmd_tag_add(args: argparse.Namespace) -> int:
name = args.name.strip()
try:
_validate_tags([name])
except ValueError as exc:
print(str(exc), file=sys.stderr)
return 1
with _connect() as conn:
exists = conn.execute("SELECT 1 FROM tags WHERE name = ?", (name,)).fetchone()
if exists:
print(f"Tag '#{name}' already exists.")
return 0
created_at = datetime.now(timezone.utc).isoformat()
conn.execute("INSERT INTO tags(name, created_at) VALUES(?, ?)", (name, created_at))
conn.commit()
_log("TAG-ADD", f"name={name}")
print(f"Tag created: #{name}")
return 0
def cmd_tag_list(args: argparse.Namespace) -> int:
with _connect() as conn:
rows = conn.execute("SELECT name FROM tags ORDER BY name").fetchall()
_log("TAG-LIST", f"returned={len(rows)}")
if not rows:
print("No tags.")
return 0
for row in rows:
print(f"#{row['name']}")
return 0
def _main() -> int:
parser = argparse.ArgumentParser(description="Note store")
sub = parser.add_subparsers(dest="cmd", required=True)
p_add = sub.add_parser("add", help="Add a note")
p_add.add_argument("text", help="Note content")
p_add.add_argument("--tags", nargs="+", metavar="TAG", default=[], help="Tags (lowercase, hyphens allowed)")
p_list = sub.add_parser("list", help="List active notes")
p_list.add_argument("--limit", type=int, default=50)
p_list.add_argument("--offset", type=int, default=0)
p_list.add_argument("--tag", nargs="+", metavar="TAG", help="Filter by tag (OR logic)")
p_show = sub.add_parser("show", help="Show one note in full by display ID")
p_show.add_argument("id", type=int, help="Display ID")
p_del = sub.add_parser("delete", help="Soft-delete a note by ID")
p_del.add_argument("id", type=int, help="Note ID")
p_tag_add = sub.add_parser("tag-add", help="Register a tag")
p_tag_add.add_argument("name", help="Tag name (lowercase, hyphens allowed)")
sub.add_parser("tag-list", help="List registered tags")
args = parser.parse_args()
if args.cmd == "add":
return cmd_add(args)
if args.cmd == "list":
return cmd_list(args)
if args.cmd == "show":
return cmd_show(args)
if args.cmd == "delete":
return cmd_delete(args)
if args.cmd == "tag-add":
return cmd_tag_add(args)
if args.cmd == "tag-list":
return cmd_tag_list(args)
return 0
if __name__ == "__main__":
sys.exit(_main())

111
skills/note/scripts/note_capture.py Normal file
View File

@@ -0,0 +1,111 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""note_capture.py — dumb, instant capture for the /note skill.
Writes the raw input verbatim into notes/inbox/ (atomic tmp -> os.replace) plus one
audit line to log/note.log, then prints a one-line confirmation. No reformulation,
no reading of the knowledge doc, no compile — that is the compile step's job
(inline in immediate mode, or the cron drain in `cron` mode).
Used identically by both modes; the only difference is what the agent does *after*
calling this (immediate: run the compile workflow inline; cron: stop).
"""
import argparse
import re
import sys
import unicodedata
from datetime import datetime
from pathlib import Path
# workspace/skills/note/scripts/note_capture.py -> parents[3] = workspace root.
WORKSPACE = Path(__file__).resolve().parents[3]
INBOX = WORKSPACE / "notes" / "inbox"
LOG = WORKSPACE / "log" / "note.log"
_SLUG_STRIP_RE = re.compile(r"[^a-z0-9]+")
_URL_RE = re.compile(r"https?://([^/\s]+)")
MAX_SLUG_WORDS = 4
SLUG_MAX_LEN = 40
def _ascii_fold(text: str) -> str:
"""Drop diacritics so Czech words survive slugging (mazání -> mazani)."""
return unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode("ascii")
def _slugify(text: str) -> str:
"""Short kebab slug from the first words of the input (domain for a bare URL)."""
first_line = next((line for line in text.splitlines() if line.strip()), "").strip()
url_match = _URL_RE.match(first_line)
if url_match:
host = url_match.group(1).removeprefix("www.")
slug = _SLUG_STRIP_RE.sub("-", _ascii_fold(host).lower()).strip("-")
return slug or "note"
words = first_line.split()[:MAX_SLUG_WORDS]
slug = _SLUG_STRIP_RE.sub("-", _ascii_fold(" ".join(words)).lower()).strip("-")
return slug[:SLUG_MAX_LEN].strip("-") or "note"
def _build_content(
captured_at: str, channel: str | None, chat_id: str | None, body: str
) -> str:
lines = [f"captured_at: {captured_at}"]
if channel:
lines.append(f"channel: {channel}")
if chat_id:
lines.append(f'chat_id: "{chat_id}"')
frontmatter = "\n".join(lines)
return f"---\n{frontmatter}\n---\n\n{body.strip()}\n"
def _append_log(filename: str, body: str) -> None:
LOG.parent.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().astimezone().isoformat(timespec="seconds")
summary = " ".join(body.split())[:80]
with LOG.open("a", encoding="utf-8") as handle:
handle.write(f"{stamp} CAPTURE {filename} :: {summary}\n")
def main() -> int:
parser = argparse.ArgumentParser(description="Capture a raw note into notes/inbox/")
parser.add_argument(
"--text", default=None, help="Raw input; if omitted, read from stdin"
)
parser.add_argument(
"--channel", default=None, help="Origin channel (telegram/websocket/cli)"
)
parser.add_argument(
"--chat-id", default=None, dest="chat_id", help="Origin chat id"
)
args = parser.parse_args()
body = args.text if args.text is not None else sys.stdin.read()
body = body.strip()
if not body:
print("Nothing to capture (empty input).", file=sys.stderr)
return 1
now = datetime.now().astimezone()
timestamp = now.strftime("%Y-%m-%d_%H_%M_%S_%f")
filename = f"{timestamp}-{_slugify(body)}.md"
content = _build_content(
now.isoformat(timespec="seconds"), args.channel, args.chat_id, body
)
INBOX.mkdir(parents=True, exist_ok=True)
tmp_path = INBOX / f".{filename}.tmp"
final_path = INBOX / filename
tmp_path.write_text(content, encoding="utf-8")
tmp_path.replace(final_path)
_append_log(filename, body)
print(f"captured: {filename}")
return 0
if __name__ == "__main__":
sys.exit(main())

216
skills/note/scripts/note_compile.py Normal file
View File

@@ -0,0 +1,216 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["nanobot-ai"]
# ///
"""note_compile.py — drain notes/inbox/ into the structured doc notes/notes.md.
Thin launcher run by the nanobot user crontab every minute. All the intelligence
lives in DRAIN_GOAL + the note skill's compile workflow; this script only decides
*when* to run and guards against concurrent runs.
Flow:
1. Cheap fs pre-check (no LLM): are there pending files in notes/inbox/? None ->
exit 0 without importing nanobot (per-minute polling stays nearly free).
2. Lockfile (notes/.compile.lock, PID + start-timestamp): another compile running?
-> exit 0. Stale lock (dead PID / older than STALE_SECONDS) is reclaimed.
3. Otherwise Nanobot.from_config().run(<drain goal>) — drains ALL pending in one
batch. process_direct has NO cron preamble (unlike cron/jobs.json agent jobs).
4. Quietly append to log/note_compile_cron.log; no Telegram.
The immediate `/note` mode runs the SAME compile workflow inline and takes the SAME
lock, so an inline merge and a background drain cannot corrupt notes.md at once.
"""
import asyncio
import json
import os
import subprocess
import sys
import traceback
from datetime import datetime
from pathlib import Path
# workspace/skills/note/scripts/note_compile.py -> parents[3] = workspace root.
WORKSPACE = Path(__file__).resolve().parents[3]
NOTES = WORKSPACE / "notes"
INBOX = NOTES / "inbox"
LOCK = NOTES / ".compile.lock"
LOG = WORKSPACE / "log" / "note_compile_cron.log"
TIMEOUT_SECONDS = 15 * 60
STALE_SECONDS = 30 * 60
DRAIN_GOAL = (
"Pomocí skillu note (Compile/drain) zpracuj VŠECHNY čekající soubory v `notes/inbox/` "
"(regulérní soubory přímo v `notes/inbox/`, mimo skryté). Pro každý postupuj podle "
"*Compile workflow* v note SKILL.md: přeformuluj na terse fakt(a) (zachovej jazyk vstupu, "
"jeden koncept per záznam, zahoď filler); z těla vytáhni VŠECHNY URL (0..N) a každou stáhni "
"přes `web` tool — když je za paywallem / login-wallem / neúplná, NEfabrikuj shrnutí, zapiš "
"jen URL + titulek + značku `⚠ paywall/neúplné`. Zařaď obsah pod správnou tematickou sekci "
"v `notes/notes.md` (novou sekci ## založ, když chybí; existující sekci uprav chirurgicky, "
"nepřepisuj celý dokument). Po úspěšném zařazení přesuň zdrojový soubor do `notes/done/`; "
"když z něj nešlo nic použitelného získat (vše za paywallem / nečitelné / nejednoznačné), "
"přesuň ho do `notes/hard/`. Přesouvej HNED po každém souboru, ať ho příští cron tik "
"nezpracovává znovu. Běžíš v izolované session na pozadí, bez interakce s uživatelem."
)
def log(message: str) -> None:
LOG.parent.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().astimezone().isoformat(timespec="seconds")
with LOG.open("a", encoding="utf-8") as handle:
handle.write(f"{stamp} {message}\n")
def pending_sources() -> list[Path]:
"""Regular files directly in notes/inbox/ (hidden files excluded; done/ and hard/ are siblings)."""
if not INBOX.exists():
return []
return [
p for p in sorted(INBOX.iterdir()) if p.is_file() and not p.name.startswith(".")
]
def _pid_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def _lock_is_stale() -> bool:
"""A lock is dead if unreadable, its PID is gone, or it is older than STALE_SECONDS."""
try:
data = json.loads(LOCK.read_text())
pid = int(data["pid"])
started = datetime.fromisoformat(data["started"])
except (OSError, ValueError, KeyError):
return True
if not _pid_alive(pid):
return True
age = (datetime.now().astimezone() - started).total_seconds()
return age > STALE_SECONDS
def acquire_lock() -> bool:
"""Atomically create the lock. Return False when a live compile already runs."""
for _ in range(2):
try:
fd = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
except FileExistsError:
if not _lock_is_stale():
return False
log("stale lock, reclaiming")
LOCK.unlink(missing_ok=True)
continue
payload = {
"pid": os.getpid(),
"started": datetime.now().astimezone().isoformat(),
}
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(payload, handle)
return True
return False
async def run_compile(goal: str) -> str:
# Heavy import deferred: the per-minute pre-check (no pending work) must not pay the
# nanobot import cost — only an actual compile run needs it.
from nanobot import Nanobot
bot = Nanobot.from_config()
result = await bot.run(goal, session_key="note-compile")
return result.content or ""
def commit_notes(count: int) -> None:
"""Stage and commit only notes/ after a successful drain.
The Dream processor owns the rest of the workspace, so we never `git add -A`.
A no-op when notes/ has no changes. .compile.lock is gitignored, so `git add notes/`
(run while the lock is still held) does not stage it. Commit failure is logged, not
raised — the drain itself already succeeded and must not be reported as failed.
"""
try:
status = subprocess.run(
["git", "-C", str(WORKSPACE), "status", "--porcelain", "notes/"],
check=True,
capture_output=True,
text=True,
)
if not status.stdout.strip():
return
subprocess.run(
["git", "-C", str(WORKSPACE), "add", "notes/"],
check=True,
capture_output=True,
text=True,
)
subprocess.run(
[
"git",
"-C",
str(WORKSPACE),
"commit",
"-m",
f"note: cron drain ({count} captures)",
],
check=True,
capture_output=True,
text=True,
)
log(f"COMMIT notes/ ({count} captures)")
except (OSError, subprocess.CalledProcessError) as error:
log(f"WARN commit failed: {error}")
def main() -> int:
dry_run = "--dry-run" in sys.argv[1:]
pending = pending_sources()
if not pending:
return 0
if not acquire_lock():
log(f"SKIP compile already running ({len(pending)} pending)")
return 0
if dry_run:
names = ", ".join(p.name for p in pending)
log(f"DRY-RUN would compile {len(pending)} pending: {names}")
LOCK.unlink(missing_ok=True)
return 0
started = datetime.now().astimezone()
log(f"START compile {len(pending)} pending: {', '.join(p.name for p in pending)}")
try:
result_text = asyncio.run(
asyncio.wait_for(run_compile(DRAIN_GOAL), timeout=TIMEOUT_SECONDS)
)
summary = (
result_text.strip().splitlines()[0][:200]
if result_text.strip()
else "(prázdný výstup)"
)
duration = int((datetime.now().astimezone() - started).total_seconds())
log(
f"END compile duration={duration}s remaining={len(pending_sources())} :: {summary}"
)
commit_notes(len(pending))
return 0
except asyncio.TimeoutError:
log(f"TIMEOUT compile po {TIMEOUT_SECONDS // 60} min")
return 1
except Exception as error:
log(f"EXCEPTION compile: {error}\n{traceback.format_exc()}")
return 1
finally:
LOCK.unlink(missing_ok=True)
if __name__ == "__main__":
sys.exit(main())

98
skills/note/tests/test_note_capture.py Normal file
View File

@@ -0,0 +1,98 @@
"""Tests for note_capture.py — dumb capture into notes/inbox/ + audit log."""
import re
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
import note_capture # noqa: E402
FILENAME_RE = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}_\d{2}_\d{2}_\d{6}-[a-z0-9-]+\.md$")
@pytest.fixture
def workspace(tmp_path, monkeypatch):
inbox = tmp_path / "notes" / "inbox"
log = tmp_path / "log" / "note.log"
monkeypatch.setattr(note_capture, "INBOX", inbox)
monkeypatch.setattr(note_capture, "LOG", log)
return tmp_path
def _run(text=None, argv_extra=None):
argv = ["note_capture.py"]
if text is not None:
argv += ["--text", text]
if argv_extra:
argv += argv_extra
return argv
def test_slugify_from_words():
assert note_capture._slugify("Mazání starých images") == "mazani-starych-images"
def test_slugify_caps_word_count():
assert note_capture._slugify("one two three four five six") == "one-two-three-four"
def test_slugify_bare_url_uses_domain():
assert (
note_capture._slugify("https://www.darkove-sklo.com/x/y") == "darkove-sklo-com"
)
def test_slugify_empty_falls_back():
assert note_capture._slugify("!!!") == "note"
def test_capture_writes_inbox_file_and_log(workspace, monkeypatch):
monkeypatch.setattr(
sys,
"argv",
_run("koupit kanistr na vodu", ["--channel", "telegram", "--chat-id", "42"]),
)
assert note_capture.main() == 0
files = list((workspace / "notes" / "inbox").glob("*.md"))
assert len(files) == 1
name = files[0].name
assert FILENAME_RE.match(name), name
content = files[0].read_text(encoding="utf-8")
assert content.startswith("---\n")
assert "channel: telegram" in content
assert 'chat_id: "42"' in content
assert "koupit kanistr na vodu" in content
log_lines = (
(workspace / "log" / "note.log").read_text(encoding="utf-8").splitlines()
)
assert len(log_lines) == 1
assert "CAPTURE" in log_lines[0]
assert name in log_lines[0]
def test_capture_no_leftover_tmp_files(workspace, monkeypatch):
monkeypatch.setattr(sys, "argv", _run("neco"))
assert note_capture.main() == 0
tmp_files = list((workspace / "notes" / "inbox").glob(".*"))
assert tmp_files == []
def test_capture_omits_absent_provenance(workspace, monkeypatch):
monkeypatch.setattr(sys, "argv", _run("bez kanalu"))
assert note_capture.main() == 0
content = next((workspace / "notes" / "inbox").glob("*.md")).read_text(
encoding="utf-8"
)
assert "channel:" not in content
assert "chat_id:" not in content
def test_capture_empty_input_fails(workspace, monkeypatch):
monkeypatch.setattr(sys, "argv", _run(" "))
assert note_capture.main() == 1
assert list((workspace / "notes" / "inbox").glob("*.md")) == []

117
skills/note/tests/test_note_compile.py Normal file
View File

@@ -0,0 +1,117 @@
"""Tests for note_compile.py — the thin drain launcher (pre-check + lock).
The nanobot import is deferred inside run_compile, so importing the module and testing
pending_sources / the lock needs no nanobot-ai install.
"""
import subprocess
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
import note_compile # noqa: E402
@pytest.fixture
def workspace(tmp_path, monkeypatch):
notes = tmp_path / "notes"
inbox = notes / "inbox"
inbox.mkdir(parents=True)
monkeypatch.setattr(note_compile, "NOTES", notes)
monkeypatch.setattr(note_compile, "INBOX", inbox)
monkeypatch.setattr(note_compile, "LOCK", notes / ".compile.lock")
monkeypatch.setattr(note_compile, "LOG", tmp_path / "log" / "note_compile_cron.log")
monkeypatch.setattr(note_compile, "WORKSPACE", tmp_path)
return tmp_path
def _git(workspace, *args) -> str:
result = subprocess.run(
["git", "-C", str(workspace), *args], check=True, capture_output=True, text=True
)
return result.stdout
def _init_repo(workspace) -> None:
_git(workspace, "init", "-q")
_git(workspace, "config", "user.email", "test@example.com")
_git(workspace, "config", "user.name", "test")
def test_pending_sources_empty(workspace):
assert note_compile.pending_sources() == []
def test_pending_sources_ignores_hidden(workspace):
inbox = workspace / "notes" / "inbox"
(inbox / "2026-07-01_10_00_00_000000-a.md").write_text("x")
(inbox / ".hidden.tmp").write_text("x")
names = [p.name for p in note_compile.pending_sources()]
assert names == ["2026-07-01_10_00_00_000000-a.md"]
def test_pending_sources_sorted(workspace):
inbox = workspace / "notes" / "inbox"
for stem in ("2026-07-01_10_00_02_000000-c", "2026-07-01_10_00_01_000000-b"):
(inbox / f"{stem}.md").write_text("x")
names = [p.name for p in note_compile.pending_sources()]
assert names == sorted(names)
def test_lock_acquire_then_blocked(workspace):
assert note_compile.acquire_lock() is True
assert note_compile.acquire_lock() is False # live lock held by this pid
def test_lock_reclaims_dead_pid(workspace):
lock = workspace / "notes" / ".compile.lock"
lock.write_text('{"pid": 999999, "started": "2999-01-01T00:00:00+00:00"}')
assert note_compile.acquire_lock() is True # dead pid -> reclaimed
def test_lock_reclaims_unparseable(workspace):
(workspace / "notes" / ".compile.lock").write_text("garbage")
assert note_compile.acquire_lock() is True
def test_main_empty_inbox_returns_zero_without_lock(workspace):
# No pending work -> exit before acquire_lock / nanobot import.
assert note_compile.main() == 0
assert not (workspace / "notes" / ".compile.lock").exists()
def test_main_dry_run_releases_lock(workspace, monkeypatch):
(workspace / "notes" / "inbox" / "2026-07-01_10_00_00_000000-a.md").write_text("x")
monkeypatch.setattr(sys, "argv", ["note_compile.py", "--dry-run"])
assert note_compile.main() == 0
assert not (workspace / "notes" / ".compile.lock").exists()
def test_commit_notes_creates_commit(workspace):
_init_repo(workspace)
(workspace / "notes" / "notes.md").write_text("# Notes\n\n## X\n\n- a fact\n")
note_compile.commit_notes(2)
assert "note: cron drain (2 captures)" in _git(workspace, "log", "--oneline")
assert "notes/notes.md" in _git(workspace, "ls-files", "notes/")
def test_commit_notes_noop_when_clean(workspace):
_init_repo(workspace)
(workspace / "notes" / "notes.md").write_text("# Notes\n")
note_compile.commit_notes(1)
before = _git(workspace, "rev-list", "--count", "HEAD").strip()
note_compile.commit_notes(1) # nothing changed since last commit
assert _git(workspace, "rev-list", "--count", "HEAD").strip() == before
def test_commit_notes_skips_gitignored_lock(workspace):
_init_repo(workspace)
(workspace / ".gitignore").write_text("notes/.compile.lock\n")
(workspace / "notes" / ".compile.lock").write_text('{"pid": 1}')
(workspace / "notes" / "notes.md").write_text("# Notes\n")
note_compile.commit_notes(1)
tracked = _git(workspace, "ls-files", "notes/")
assert "notes/notes.md" in tracked
assert ".compile.lock" not in tracked