Hands-On: Build an Idempotent Intake Manifest
While writing Deterministic First: Building a Knowledge Intake Pipeline, I kept coming back to the same problem: a successful first import does not prove very much. Plenty of scripts can create a row once. I wanted to know what happened when I ran the script again, changed a file, removed it, and then put it back. If the inventory could not explain each of those events without producing duplicates or pretending stale data was still current, I was not ready to build anything else on top of it.
That led me to a small standard-library Python utility backed by SQLite. It scans one deliberately bounded directory tree, records deterministic source identities, preserves the history of missing files, migrates the earlier teaching schema, and reports what it did as JSON. I stopped it there on purpose. It does not summarize a document, classify it, approve it, or decide whether an agent may retrieve it. I want those later decisions to remain visible instead of disappearing inside an importer that tries to do everything.
That configured root is an allowlist boundary, not an invitation to inventory every note the agent can access. I would point this utility at a deliberately selected source tree and send its output into review; I would not aim it at an entire synchronized Main Vault by default.
The lab has two files: intake_manifest.py contains the utility, and test_intake_manifest.py contains eight canaries. Everything runs with the Python standard library. The sqlite3 command-line client is useful for the inspection step, but the manifest itself does not require it.
Before You Start
Use a shell with Python 3 and curl. The walkthrough builds everything under /tmp/intake-demo, so it does not need access to your notes, Vault, agent workspace, or an existing database. If that path already contains something you need, choose another disposable directory before running the commands.
| Stage | What you will prove |
|---|---|
| Define the contract | Every field answers a lifecycle or provenance question |
| Run the first scan | One allowlisted source creates one durable identity |
| Rerun and change it | Unchanged content stays stable and changed content re-enters processing |
| Remove and restore it | Missing history survives and restoration keeps the original identity |
| Inspect and test it | SQLite state and eight canaries agree with the command output |
| Hand it off | Review and retrieval eligibility remain separate decisions |
Step 1: Define the Manifest Contract
I started with one row for every allowlisted Markdown or text file. There is nothing particularly glamorous in the record, but every field answers a question I know I will have to ask later:
| Field | What it tells me |
|---|---|
source_id |
A stable identity derived from source type and canonical path |
source_path |
Where the source authority was observed |
content_sha256 |
Whether the bytes actually changed |
size_bytes, mtime_ns |
Observed metadata, refreshed even when content is unchanged |
extractor_version |
Which inventory contract evaluated the source |
processing_state |
Whether downstream processing must reconsider it |
availability_state |
Whether the source is currently present or retained as missing |
discovered_at |
When the manifest first observed the source |
content_changed_at |
When its content or extractor contract last changed |
updated_at, last_seen_at |
When manifest state changed and when a completed scan last saw it |
I tied identity to the source type and canonical path because those are the facts this little utility can actually prove. If I rename a file, the old path becomes missing and the new path becomes a new source. That may be less clever than guessing that the two files are the same thing, but it is also auditable. In a larger system I would record a known move explicitly or use a stronger identifier supplied by the source system.
Once I wrote those rules down, the control flow became almost boring, which is exactly what I wanted. One transaction selects eligible files beneath the configured root, fingerprints each stable read, compares it with the previous state, reconciles missing records within that same root, checks SQLite integrity, and only then commits.
I did not want the convenience of a recursive scan to turn into accidental access to an entire home directory. Hidden paths, non-text suffixes, and file or directory symlinks that could escape the root are excluded. The utility also stats each selected file before and after hashing it. If the file changes during the read, the scan fails and rolls back instead of committing a fingerprint assembled from two different versions.
Step 2: Get the Complete Lab
I am including both complete files because sample code is much more useful when the reader can run the same checks I ran. The on-page source viewer and download links use the same public assets, so there is no second hand-maintained copy waiting to drift away from the article. The first disclosure contains the utility, and the second contains its eight-canary acceptance suite.
intake_manifest.py python View source
#!/usr/bin/env python3
"""Maintain a deterministic SQLite inventory for Markdown and text files."""
from __future__ import annotations
import argparse
import hashlib
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
VERSION = "intake-manifest-v2"
SCHEMA_VERSION = 2
ALLOWED_SUFFIXES = {".md", ".txt"}
SCHEMA = """
CREATE TABLE IF NOT EXISTS sources (
source_id TEXT PRIMARY KEY,
source_path TEXT NOT NULL UNIQUE,
source_type TEXT NOT NULL,
content_sha256 TEXT NOT NULL,
size_bytes INTEGER NOT NULL,
mtime_ns INTEGER NOT NULL,
extractor_version TEXT NOT NULL,
processing_state TEXT NOT NULL,
availability_state TEXT NOT NULL,
discovered_at TEXT NOT NULL,
content_changed_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
last_seen_at TEXT NOT NULL
);
"""
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
def stable_id(source_type: str, path: Path) -> str:
"""Return an identity stable for one source type and canonical path."""
identity = f"{source_type}:{path.resolve()}"
return "src_" + hashlib.sha256(identity.encode()).hexdigest()[:24]
def file_fingerprint(path: Path) -> tuple[str, Any]:
"""Hash a file and refuse a result if it changed during the read."""
before = path.stat()
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
after = path.stat()
if (before.st_size, before.st_mtime_ns) != (after.st_size, after.st_mtime_ns):
raise RuntimeError(f"source changed while being fingerprinted: {path}")
return digest.hexdigest(), after
def source_files(root: Path) -> list[Path]:
"""Return allowlisted regular files without following file symlinks."""
return [
path
for path in sorted(root.rglob("*"))
if path.is_file()
and not path.is_symlink()
and is_below(path, root)
and path.suffix.lower() in ALLOWED_SUFFIXES
and not any(part.startswith(".") for part in path.relative_to(root).parts)
]
def is_below(path: Path, root: Path) -> bool:
try:
path.resolve().relative_to(root)
except ValueError:
return False
return True
def ensure_schema(db: sqlite3.Connection) -> None:
"""Create v2 or migrate the original teaching schema without data loss."""
db.execute(SCHEMA)
columns = {
str(row[1]) for row in db.execute("PRAGMA table_info(sources)").fetchall()
}
additions = {
"availability_state": "TEXT NOT NULL DEFAULT 'present'",
"content_changed_at": "TEXT NOT NULL DEFAULT ''",
"last_seen_at": "TEXT NOT NULL DEFAULT ''",
}
for name, definition in additions.items():
if name not in columns:
db.execute(f"ALTER TABLE sources ADD COLUMN {name} {definition}")
db.execute(
"UPDATE sources SET content_changed_at=updated_at WHERE content_changed_at=''"
)
db.execute("UPDATE sources SET last_seen_at=updated_at WHERE last_seen_at=''")
db.execute(f"PRAGMA user_version={SCHEMA_VERSION}")
db.commit()
def manifest_records(db: sqlite3.Connection, root: Path) -> list[dict[str, Any]]:
rows = db.execute(
"""
SELECT source_id, source_path, source_type, content_sha256,
size_bytes, mtime_ns, extractor_version, processing_state,
availability_state, discovered_at, content_changed_at,
updated_at, last_seen_at
FROM sources
ORDER BY source_path
"""
).fetchall()
return [dict(row) for row in rows if is_below(Path(row["source_path"]), root)]
def inventory(
root: Path,
database: Path,
*,
extractor_version: str = VERSION,
include_records: bool = False,
) -> dict[str, Any]:
root = root.resolve()
database = database.resolve()
if not root.is_dir():
raise SystemExit(f"source directory does not exist: {root}")
database.parent.mkdir(parents=True, exist_ok=True)
counts = {
"new": 0,
"changed": 0,
"unchanged": 0,
"restored": 0,
"missing": 0,
}
with sqlite3.connect(database) as db:
db.row_factory = sqlite3.Row
ensure_schema(db)
db.execute("BEGIN IMMEDIATE")
now = utc_now()
seen: set[str] = set()
try:
for path in source_files(root):
absolute = path.resolve()
source_type = path.suffix.lower().lstrip(".")
source_id = stable_id(source_type, absolute)
digest, stat = file_fingerprint(absolute)
seen.add(source_id)
prior = db.execute(
"""
SELECT content_sha256, extractor_version, availability_state
FROM sources WHERE source_id=?
""",
(source_id,),
).fetchone()
if prior is None:
counts["new"] += 1
db.execute(
"""
INSERT INTO sources (
source_id, source_path, source_type, content_sha256,
size_bytes, mtime_ns, extractor_version,
processing_state, availability_state, discovered_at,
content_changed_at, updated_at, last_seen_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 'discovered', 'present',
?, ?, ?, ?)
""",
(
source_id,
str(absolute),
source_type,
digest,
stat.st_size,
stat.st_mtime_ns,
extractor_version,
now,
now,
now,
now,
),
)
continue
same_content = prior["content_sha256"] == digest
same_version = prior["extractor_version"] == extractor_version
was_missing = prior["availability_state"] == "missing"
if same_content and same_version and not was_missing:
counts["unchanged"] += 1
db.execute(
"""
UPDATE sources
SET source_path=?, size_bytes=?, mtime_ns=?,
availability_state='present', last_seen_at=?
WHERE source_id=?
""",
(
str(absolute),
stat.st_size,
stat.st_mtime_ns,
now,
source_id,
),
)
continue
if same_content and same_version:
counts["restored"] += 1
db.execute(
"""
UPDATE sources
SET source_path=?, source_type=?, size_bytes=?, mtime_ns=?,
processing_state='discovered',
availability_state='present', updated_at=?, last_seen_at=?
WHERE source_id=?
""",
(
str(absolute),
source_type,
stat.st_size,
stat.st_mtime_ns,
now,
now,
source_id,
),
)
continue
counts["changed"] += 1
db.execute(
"""
UPDATE sources
SET source_path=?, source_type=?, content_sha256=?,
size_bytes=?, mtime_ns=?, extractor_version=?,
processing_state='discovered', availability_state='present',
content_changed_at=?, updated_at=?, last_seen_at=?
WHERE source_id=?
""",
(
str(absolute),
source_type,
digest,
stat.st_size,
stat.st_mtime_ns,
extractor_version,
now,
now,
now,
source_id,
),
)
prior_rows = db.execute(
"SELECT source_id, source_path, availability_state FROM sources"
).fetchall()
for row in prior_rows:
if (
row["source_id"] not in seen
and row["availability_state"] != "missing"
and is_below(Path(row["source_path"]), root)
):
counts["missing"] += 1
db.execute(
"""
UPDATE sources
SET availability_state='missing', updated_at=?
WHERE source_id=?
""",
(now, row["source_id"]),
)
check = db.execute("PRAGMA quick_check").fetchone()[0]
if check != "ok":
raise RuntimeError(f"manifest integrity check failed: {check}")
db.commit()
except Exception:
db.rollback()
raise
result: dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"extractor_version": extractor_version,
**counts,
}
if include_records:
result["records"] = manifest_records(db, root)
return result
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("source_directory", type=Path)
parser.add_argument("--db", required=True, type=Path, help="SQLite manifest path")
parser.add_argument(
"--records",
action="store_true",
help="include ordered records for this source root in the JSON result",
)
args = parser.parse_args()
print(
json.dumps(
inventory(args.source_directory, args.db, include_records=args.records),
indent=2,
sort_keys=True,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
test_intake_manifest.py python View source
#!/usr/bin/env python3
"""Canary tests for the Post 2A intake manifest utility."""
from __future__ import annotations
import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest import mock
import intake_manifest
from intake_manifest import inventory
class IntakeManifestTests(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name) / "sources"
self.root.mkdir()
self.database = Path(self.temporary.name) / "manifest.db"
def tearDown(self) -> None:
self.temporary.cleanup()
def write(self, relative: str, content: str) -> Path:
path = self.root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
return path
def test_new_unchanged_and_changed_are_one_durable_record(self) -> None:
source = self.write("first.md", "first version\n")
first = inventory(self.root, self.database, include_records=True)
second = inventory(self.root, self.database, include_records=True)
source.write_text("second version\n")
third = inventory(self.root, self.database, include_records=True)
self.assertEqual(first["new"], 1)
self.assertEqual(second["unchanged"], 1)
self.assertEqual(third["changed"], 1)
self.assertEqual(len(third["records"]), 1)
self.assertEqual(
first["records"][0]["source_id"], third["records"][0]["source_id"]
)
def test_missing_source_is_retained_and_can_be_restored(self) -> None:
source = self.write("first.md", "retained history\n")
first = inventory(self.root, self.database, include_records=True)
source.unlink()
missing = inventory(self.root, self.database, include_records=True)
source.write_text("retained history\n")
restored = inventory(self.root, self.database, include_records=True)
self.assertEqual(missing["missing"], 1)
self.assertEqual(missing["records"][0]["availability_state"], "missing")
self.assertEqual(restored["restored"], 1)
self.assertEqual(restored["records"][0]["availability_state"], "present")
self.assertEqual(
restored["records"][0]["content_changed_at"],
first["records"][0]["content_changed_at"],
)
def test_hidden_non_text_and_symlink_sources_are_not_ingested(self) -> None:
self.write("visible.md", "included\n")
self.write(".hidden.md", "excluded\n")
self.write("image.png", "not text\n")
outside = Path(self.temporary.name) / "outside.md"
outside.write_text("outside root\n")
(self.root / "linked.md").symlink_to(outside)
outside_directory = Path(self.temporary.name) / "outside-directory"
outside_directory.mkdir()
(outside_directory / "nested.md").write_text("also outside root\n")
(self.root / "linked-directory").symlink_to(outside_directory)
result = inventory(self.root, self.database, include_records=True)
self.assertEqual(result["new"], 1)
self.assertEqual(
[Path(record["source_path"]).name for record in result["records"]],
["visible.md"],
)
def test_extractor_upgrade_resets_processing_state(self) -> None:
self.write("first.md", "same content\n")
inventory(self.root, self.database, extractor_version="extractor-v1")
with sqlite3.connect(self.database) as db:
db.execute("UPDATE sources SET processing_state='processed'")
db.commit()
result = inventory(
self.root,
self.database,
extractor_version="extractor-v2",
include_records=True,
)
self.assertEqual(result["changed"], 1)
self.assertEqual(result["records"][0]["processing_state"], "discovered")
def test_missing_reconciliation_is_scoped_to_the_scanned_root(self) -> None:
self.write("first.md", "first root\n")
second_root = Path(self.temporary.name) / "other-sources"
second_root.mkdir()
(second_root / "second.md").write_text("second root\n")
inventory(self.root, self.database)
inventory(second_root, self.database)
result = inventory(self.root, self.database)
self.assertEqual(result["missing"], 0)
with sqlite3.connect(self.database) as db:
states = dict(
db.execute(
"SELECT source_path, availability_state FROM sources"
).fetchall()
)
self.assertEqual(states[str((second_root / "second.md").resolve())], "present")
def test_unchanged_scan_refreshes_observed_metadata(self) -> None:
source = self.write("first.md", "same content\n")
first = inventory(self.root, self.database, include_records=True)
source.touch()
second = inventory(self.root, self.database, include_records=True)
self.assertEqual(second["unchanged"], 1)
self.assertGreaterEqual(
second["records"][0]["mtime_ns"], first["records"][0]["mtime_ns"]
)
self.assertGreaterEqual(
second["records"][0]["last_seen_at"],
first["records"][0]["last_seen_at"],
)
def test_failed_scan_rolls_back_all_source_changes(self) -> None:
self.write("first.md", "one\n")
self.write("second.md", "two\n")
real_fingerprint = intake_manifest.file_fingerprint
calls = 0
def fail_second(path: Path):
nonlocal calls
calls += 1
if calls == 2:
raise RuntimeError("simulated unstable source")
return real_fingerprint(path)
with mock.patch.object(intake_manifest, "file_fingerprint", fail_second):
with self.assertRaisesRegex(RuntimeError, "simulated unstable source"):
inventory(self.root, self.database)
with sqlite3.connect(self.database) as db:
count = db.execute("SELECT COUNT(*) FROM sources").fetchone()[0]
self.assertEqual(count, 0)
def test_v1_database_is_migrated_without_losing_rows(self) -> None:
source = self.write("first.md", "migrated\n")
with sqlite3.connect(self.database) as db:
db.execute(
"""
CREATE TABLE sources (
source_id TEXT PRIMARY KEY,
source_path TEXT NOT NULL UNIQUE,
source_type TEXT NOT NULL,
content_sha256 TEXT NOT NULL,
size_bytes INTEGER NOT NULL,
mtime_ns INTEGER NOT NULL,
extractor_version TEXT NOT NULL,
processing_state TEXT NOT NULL,
discovered_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
digest, stat = intake_manifest.file_fingerprint(source)
db.execute(
"""
INSERT INTO sources VALUES (?, ?, 'md', ?, ?, ?, ?, 'processed', ?, ?)
""",
(
intake_manifest.stable_id("md", source),
str(source.resolve()),
digest,
stat.st_size,
stat.st_mtime_ns,
intake_manifest.VERSION,
"2026-01-01T00:00:00+00:00",
"2026-01-01T00:00:00+00:00",
),
)
db.commit()
result = inventory(self.root, self.database, include_records=True)
self.assertEqual(result["unchanged"], 1)
self.assertEqual(len(result["records"]), 1)
self.assertEqual(result["records"][0]["processing_state"], "processed")
with sqlite3.connect(self.database) as db:
schema_version = db.execute("PRAGMA user_version").fetchone()[0]
integrity = db.execute("PRAGMA quick_check").fetchone()[0]
self.assertEqual(schema_version, intake_manifest.SCHEMA_VERSION)
self.assertEqual(integrity, "ok")
if __name__ == "__main__":
unittest.main()
Step 3: Run a Clean End-to-End Scan
I use a disposable tree under /tmp so I can repeat the exercise from a known starting point. Download both files into it and create one small Markdown source:
rm -rf /tmp/intake-demo
mkdir -p /tmp/intake-demo/sources
curl -fsSLo /tmp/intake-demo/intake_manifest.py \
https://unixwzrd.ai/assets/code/agent-optimization/intake_manifest.py
curl -fsSLo /tmp/intake-demo/test_intake_manifest.py \
https://unixwzrd.ai/assets/code/agent-optimization/test_intake_manifest.py
printf '%s\n' '# First note' 'A source with enough text to fingerprint.' \
> /tmp/intake-demo/sources/first.md
python3 /tmp/intake-demo/intake_manifest.py \
/tmp/intake-demo/sources \
--db /tmp/intake-demo/manifest.db
The first result is not exciting, and that is good:
{
"changed": 0,
"extractor_version": "intake-manifest-v2",
"missing": 0,
"new": 1,
"restored": 0,
"schema_version": 2,
"unchanged": 0
}
When I run the same command again, it reports new: 0 and unchanged: 1. The program still reads and hashes the file, so I am not claiming that modification time provides a shortcut around the scan. What I am claiming is narrower and easier to verify: an ordinary rerun does not invent another identity or reset work unnecessarily.
Now change the source and ask for the ordered record set:
printf '%s\n' 'A second paragraph changes the content fingerprint.' \
>> /tmp/intake-demo/sources/first.md
python3 /tmp/intake-demo/intake_manifest.py \
/tmp/intake-demo/sources \
--db /tmp/intake-demo/manifest.db \
--records
This time the summary reports changed: 1, but the records array still contains one row with the same source_id. Its content_sha256 changes and its processing_state returns to "discovered", which is the behavior I need before handing the source to another stage. Changing the extractor version causes the same reprocessing decision even when the file itself is unchanged; I do not want a new inventory contract masquerading as the old one.
At this checkpoint, three scans should still have produced one source row. The first run proves creation, the second proves idempotence, and the third proves that content changes reset work without inventing a new identity.
Step 4: Test Disappearance and Restoration
This is where the original small example was not good enough. If a file disappeared, its old row simply survived and still looked current. Deleting the row would have hidden the opposite fact: the source had existed, and something had happened to it. The useful behavior is to retain the record while changing its availability. Move the example outside the source root and scan again:
mv /tmp/intake-demo/sources/first.md /tmp/intake-demo/first.saved
python3 /tmp/intake-demo/intake_manifest.py \
/tmp/intake-demo/sources \
--db /tmp/intake-demo/manifest.db \
--records
The result reports missing: 1, and the retained row now has availability_state: "missing". Nothing is silently erased and nothing stale is presented as current. Put the same file back and scan once more:
mv /tmp/intake-demo/first.saved /tmp/intake-demo/sources/first.md
python3 /tmp/intake-demo/intake_manifest.py \
/tmp/intake-demo/sources \
--db /tmp/intake-demo/manifest.db \
--records
Now the result reports restored: 1. The record keeps its original identity and content_changed_at value because availability changed but the bytes did not. I still return its processing state to discovered; a downstream stage should make an explicit decision about the restored authority rather than assuming nothing important happened.
Step 5: Inspect the SQLite State
The --records option is convenient for scripts, but one reason I chose SQLite is that I can inspect the durable state without standing up another service:
python3 - <<'PY'
import sqlite3
connection = sqlite3.connect("/tmp/intake-demo/manifest.db")
connection.row_factory = sqlite3.Row
rows = connection.execute(
"""SELECT source_id, availability_state, processing_state,
size_bytes, extractor_version
FROM sources
ORDER BY source_path"""
)
for row in rows:
print(dict(row))
connection.close()
PY
The manifest stores canonical paths because it is local provenance state. I would not copy those paths into a public report or diagnostic bundle, and I have not used paths from my own environment in this article. If paths are sensitive in your environment, protect the database or introduce an opaque locator at the system boundary.
Step 6: Run the Canary Suite
I also do not want the reader to trust this utility simply because the example output looks reasonable. The earlier version left too much correctness as an exercise, so this version includes the acceptance tests I use to check the boundary:
cd /tmp/intake-demo
python3 -m unittest -v test_intake_manifest.py
The eight canaries cover the failure modes I care about at this boundary:
| Canary | Evidence it demands |
|---|---|
| New, unchanged, and changed | Three scans retain one stable source row |
| Missing and restored | Disappearance retains history and restoration retains identity |
| Bounded selection | Hidden, non-text, and symlinked sources remain excluded |
| Extractor upgrade | A contract-version change resets processing to discovered |
| Root-scoped reconciliation | Scanning one configured tree cannot mark another tree’s records missing |
| Metadata refresh | Unchanged content still refreshes observed metadata |
| Transaction rollback | A failed fingerprint leaves no partial source updates |
| Schema migration | The original teaching schema upgrades without losing its row or state |
The ordering matters here. The integrity check runs before commit, so a source-read failure or failed PRAGMA quick_check rolls back the source changes from that invocation. Schema migration commits separately, which allows an older manifest to be upgraded even when a later source scan fails.
Step 7: Hand Off to Review Without Granting Retrieval
This is the point where it is tempting to keep adding features, and it is also where I stop. A downstream extractor can select rows in processing_state = "discovered", but its output should still enter review in a fail-closed state:
---
classification: "unclassified"
review_state: "pending"
mnemosyne: "exclude"
source_id: "src_example"
source_hash: "example-sha256"
classification_policy: "[[Wiki/Policies/Document Classification Policy]]"
related:
---
After a person accepts the material for durable filing, retrieval eligibility remains a separate decision:
---
classification: "public"
review_state: "approved"
mnemosyne: "exclude"
source_id: "src_example"
source_hash: "example-sha256"
classification_policy: "[[Wiki/Policies/Document Classification Policy]]"
related:
- "[[Wiki/Concepts/Example Topic]]"
---
The important line is still mnemosyne: "exclude". Filing something after review does not silently authorize an agent to retrieve it or inject it into a prompt. I retain rejected candidates too, with review_state: "rejected", mnemosyne: "exclude", and a decision timestamp, so the next extraction pass does not present the same material as new work.
What You Should Have at the End
You should now have a disposable SQLite manifest that can explain new, unchanged, changed, missing, restored, and renamed sources without losing earlier state. The command output, direct SQL inspection, and canary suite should all describe the same boundary. If they do not, stop there rather than handing the manifest to an extractor.
Optional Extensions: Try Three Useful Experiments
The tested lab is complete at this point. These three small extensions are optional, but each has a concrete result you can check.
First, add a second Markdown file and scan again:
printf '%s\n' '# Second note' 'Another allowlisted source.' \
> /tmp/intake-demo/sources/second.md
python3 /tmp/intake-demo/intake_manifest.py \
/tmp/intake-demo/sources \
--db /tmp/intake-demo/manifest.db \
--records
The summary should report new: 1 and unchanged: 1, with two present records.
Next, add a hidden Markdown file and a non-allowlisted JSON file, then scan again:
printf '%s\n' '# Hidden note' > /tmp/intake-demo/sources/.hidden.md
printf '%s\n' '{"ignored": true}' > /tmp/intake-demo/sources/ignored.json
python3 /tmp/intake-demo/intake_manifest.py \
/tmp/intake-demo/sources \
--db /tmp/intake-demo/manifest.db \
--records
The summary should report new: 0 and unchanged: 2. Neither added file should appear in records because hidden paths and non-text suffixes are outside the source contract.
Finally, rename the original source and scan once more:
mv /tmp/intake-demo/sources/first.md \
/tmp/intake-demo/sources/renamed.md
python3 /tmp/intake-demo/intake_manifest.py \
/tmp/intake-demo/sources \
--db /tmp/intake-demo/manifest.db \
--records
The summary should report new: 1, missing: 1, and unchanged: 1. The old path remains as missing history while the renamed path receives a new source identity. The utility does not guess that a rename is a move; a larger adapter would need an explicit move record or a stronger source-system identifier to establish that relationship.
When you are finished experimenting, remove only the disposable directory you created for this lab:
rm -rf /tmp/intake-demo
Current State
I am comfortable calling this utility complete for its stated job. It maintains an inspectable, transactional inventory of allowlisted local text sources and reports their lifecycle across reruns. Its JSON and SQLite contracts are suitable inputs for a separate extractor, and the included tests make that claim reproducible rather than anecdotal.
It still does not snapshot application databases, parse live session formats, classify content, remove secrets, review candidates, or populate a memory engine. Those jobs need source-specific adapters and policy decisions. Leaving them outside this program is not unfinished homework. It is how I keep one small component understandable enough to trust.
Next Work
From here I can move into memory governance without pretending the inventory solved it: who may retrieve which durable material, where authorization has to happen, and why access control must narrow the candidate set before ranking begins. The next main installment tells that part of the story, and its own Hands-On companion will turn the ordering into a small fail-closed router with canary tests.
Join the Discussion
Comments for this post live in GitHub Discussions. That keeps moderation in one place and gives the conversation a stable home.