Skip to content

[MEDIUM] perf(approve): batch approval-log and HMAC-ledger writes per commit #44

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
elhoim wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[MEDIUM] perf(approve): batch approval-log and HMAC-ledger writes per commit #44

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
af61ff3
perf(approve): batch approval-log and ledger writes per commit
elhoim Sep 16, 2026
d8a594e
Merge upstream main into perf/batch-approval-and-ledger-writes
elhoim Oct 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 49 additions & 12 deletions src/vhir_cli/case_io.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,34 +270,71 @@ def write_approval_log(
coupled_from: str = "",
) -> bool:
"""Write approval/rejection record to approvals.jsonl. Returns True on success."""
log_file = case_dir / "approvals.jsonl"
return write_approval_log_batch(
case_dir,
[
{
"item_id": item_id,
"action": action,
"identity": identity,
"reason": reason,
"mode": mode,
"content_hash": content_hash,
"stale_at_approval": stale_at_approval,
"coupled_from": coupled_from,
}
],
)


def _approval_entry(record: dict) -> dict:
"""Build one approvals.jsonl record from write_approval_log arguments."""
identity = record["identity"]
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"item_id": item_id,
"action": action,
"item_id": record["item_id"],
"action": record["action"],
"os_user": identity["os_user"],
"examiner": identity.get("examiner", identity.get("analyst", "")),
"examiner_source": identity.get(
"examiner_source", identity.get("analyst_source", "")
),
"mode": mode,
"mode": record.get("mode", "interactive"),
}
if reason:
entry["reason"] = reason
if content_hash:
entry["content_hash"] = content_hash
if stale_at_approval:
if record.get("reason"):
entry["reason"] = record["reason"]
if record.get("content_hash"):
entry["content_hash"] = record["content_hash"]
if record.get("stale_at_approval"):
entry["stale_at_approval"] = True
if coupled_from:
entry["coupled_from"] = coupled_from
if record.get("coupled_from"):
entry["coupled_from"] = record["coupled_from"]
return entry


def write_approval_log_batch(case_dir: Path, records: list[dict]) -> bool:
"""Write approval/rejection records to approvals.jsonl in one pass.

Each record carries the keyword arguments of write_approval_log except
case_dir. One unlock/open/fsync/lock cycle covers the whole batch, so a
commit pays a single durability stall instead of one per disposition.
Records are appended in list order. Returns True on success; on failure
no record in the batch is reported as written.
"""
if not records:
return True
log_file = case_dir / "approvals.jsonl"
# Serialize before touching the file so a malformed record cannot leave a
# partially written batch behind.
lines = "".join(json.dumps(_approval_entry(r)) + "\n" for r in records)
try:
if log_file.exists():
os.chmod(log_file, 0o644)
except OSError:
pass
try:
with open(log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(entry) + "\n")
f.write(lines)
f.flush()
os.fsync(f.fileno())
except OSError:
Expand Down
183 changes: 106 additions & 77 deletions src/vhir_cli/commands/approve.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
save_findings,
save_timeline,
save_todos,
write_approval_log,
write_approval_log_batch,
)


Expand Down Expand Up @@ -249,16 +249,18 @@ def _approve_specific(
# Step 2: Audit log (best-effort, warn on failure)
log_failures = []
all_approved = list(to_approve) + coupled_events
for item in all_approved:
if not write_approval_log(
case_dir,
item["id"],
"APPROVED",
identity,
mode=mode,
content_hash=item["content_hash"],
):
log_failures.append(item["id"])
log_records = [
{
"item_id": item["id"],
"action": "APPROVED",
"identity": identity,
"mode": mode,
"content_hash": item["content_hash"],
}
for item in all_approved
]
if not write_approval_log_batch(case_dir, log_records):
log_failures = [r["item_id"] for r in log_records]

# Step 3: HMAC ledger (warn on failure)
hmac_failures = _write_verification_entries(
Expand Down Expand Up @@ -507,24 +509,31 @@ def _interactive_review(

# Step 2: Approval log (best-effort, collect failures)
log_failures = []
log_records = []
for item in all_items:
if item["id"] in approvals and not write_approval_log(
case_dir,
item["id"],
"APPROVED",
identity,
mode=mode,
content_hash=item["content_hash"],
):
log_failures.append(item["id"])
if item["id"] in approvals:
log_records.append(
{
"item_id": item["id"],
"action": "APPROVED",
"identity": identity,
"mode": mode,
"content_hash": item["content_hash"],
}
)
for item in all_items:
disp = dispositions.get(item["id"])
if disp and disp[0] == "reject":
reason = disp[1] or ""
if not write_approval_log(
case_dir, item["id"], "REJECTED", identity, reason=reason, mode=mode
):
log_failures.append(item["id"])
log_records.append(
{
"item_id": item["id"],
"action": "REJECTED",
"identity": identity,
"reason": reason,
"mode": mode,
}
)
# Coupled items also need audit log entries
for item in coupled_tl:
status = item.get("status", "")
Expand All @@ -533,24 +542,31 @@ def _interactive_review(
if status == "REJECTED"
else ""
)
if not write_approval_log(
case_dir,
item["id"],
status,
identity,
mode=mode,
coupled_from=item.get("auto_created_from", ""),
content_hash=item.get("content_hash", ""),
reason=reason,
):
log_failures.append(item["id"])
log_records.append(
{
"item_id": item["id"],
"action": status,
"identity": identity,
"mode": mode,
"coupled_from": item.get("auto_created_from", ""),
"content_hash": item.get("content_hash", ""),
"reason": reason,
}
)
for item in coupled_ioc:
status = item.get("status", "")
reason = item.get("rejection_reason", "") if status == "REJECTED" else ""
if not write_approval_log(
case_dir, item["id"], status, identity, mode=mode, reason=reason
):
log_failures.append(item["id"])
log_records.append(
{
"item_id": item["id"],
"action": status,
"identity": identity,
"mode": mode,
"reason": reason,
}
)
if not write_approval_log_batch(case_dir, log_records):
log_failures = [r["item_id"] for r in log_records]

# Step 3: HMAC ledger (warn on failure)
approved_items = [item for item in all_items if item["id"] in approvals]
Expand Down Expand Up @@ -598,7 +614,7 @@ def _write_verification_entries(
from vhir_cli.verification import (
compute_hmac,
derive_hmac_key,
write_ledger_entry,
write_ledger_entries,
)
except ImportError:
return [item.get("id", "") for item in items]
Expand All @@ -625,6 +641,7 @@ def _write_verification_entries(
case_id = case_dir.name

failures: list[str] = []
entries = []
for item in items:
item_id = item.get("id", "")
item_type = (
Expand All @@ -635,20 +652,22 @@ def _write_verification_entries(
else "finding"
)
desc = hmac_text(item)
entry = {
"finding_id": item_id,
"type": item_type,
"hmac": compute_hmac(derived_key, desc),
"hmac_version": 2,
"content_snapshot": desc,
"approved_by": identity["examiner"],
"approved_at": now,
"case_id": case_id,
}
try:
write_ledger_entry(case_id, entry)
except OSError:
failures.append(item_id)
entries.append(
{
"finding_id": item_id,
"type": item_type,
"hmac": compute_hmac(derived_key, desc),
"hmac_version": 2,
"content_snapshot": desc,
"approved_by": identity["examiner"],
"approved_at": now,
"case_id": case_id,
}
)
try:
write_ledger_entries(case_id, entries)
except OSError:
failures = [entry["finding_id"] for entry in entries]

return failures

Expand Down Expand Up @@ -1339,37 +1358,47 @@ def _review_mode(case_dir: Path, identity: dict, config_path: Path) -> None:

# Step 2: Approval log (best-effort, collect failures)
log_failures = []
log_records = []
for item_id in approved_ids:
item = item_by_id.get(item_id)
if item and not write_approval_log(
case_dir,
item_id,
"APPROVED",
identity,
mode=mode,
content_hash=item.get("content_hash", ""),
stale_at_approval=item_id in stale_warnings,
):
log_failures.append(item_id)
if item:
log_records.append(
{
"item_id": item_id,
"action": "APPROVED",
"identity": identity,
"mode": mode,
"content_hash": item.get("content_hash", ""),
"stale_at_approval": item_id in stale_warnings,
}
)
for entry in rejections:
item_id = entry.get("id", "")
if item_id in rejected_ids:
reason = entry.get("rejection_reason", "") or entry.get("reason", "")
if not write_approval_log(
case_dir, item_id, "REJECTED", identity, reason=reason, mode=mode
):
log_failures.append(item_id)
log_records.append(
{
"item_id": item_id,
"action": "REJECTED",
"identity": identity,
"reason": reason,
"mode": mode,
}
)
for item_id in edited_ids:
item = item_by_id.get(item_id)
if item and not write_approval_log(
case_dir,
item_id,
"EDITED",
identity,
mode=mode,
content_hash=item.get("content_hash", ""),
):
log_failures.append(item_id)
if item:
log_records.append(
{
"item_id": item_id,
"action": "EDITED",
"identity": identity,
"mode": mode,
"content_hash": item.get("content_hash", ""),
}
)
if not write_approval_log_batch(case_dir, log_records):
log_failures = [r["item_id"] for r in log_records]

# Step 3: HMAC ledger (warn on failure)
approved_items = [item_by_id[aid] for aid in approved_ids if aid in item_by_id]
Expand Down
Loading