holdpoint

A hold point is a stage where work stops until someone authorized releases it. holdpoint gives your agent one.

The problem

Your agent drafts an email to a customer, and the draft is good — most of the time. The one time it is not, smtp.send() has already run. Every team that ships agents ends up writing the same layer by hand: park the risky call somewhere, let a human look at it, run it after approval. Hand-rolled versions tend to share the same bugs: the action runs twice when two reviewers click at once, approved work vanishes when the process restarts, and nobody can say afterwards who approved what.

holdpoint is that layer as a library: one primitive — a deferred function call waiting for a decision — with the concurrency, durability, and audit questions answered once. Zero runtime dependencies; state lives in a SQLite file you own.

Install

pip install holdpoint

Python 3.10+. Nothing else.

Example

from holdpoint import HoldPoint, SQLiteStore

hold = HoldPoint(store=SQLiteStore("approvals.db"))

@hold.guard(kind="send_email", ttl="24h")
def send_email(to: str, subject: str) -> str:
    print(f"sending {subject!r} to {to}")
    return "sent"

pending = send_email(to="ceo@corp.com", subject="Proposal")  # queued, NOT sent
print(pending.status.value)                                  # "pending"

for item in hold.pending():                                  # your dashboard, CLI, or bot
    result = hold.approve(item.id, actor="sava", patch={"subject": "Shorter"})
    print(result.value)                                      # "sent"

Calling the function stores the call. Approving executes it — once — with the reviewer's edits applied. Everything lands in an append-only audit trail in the same database file.

When you don't need this

Honesty first: holdpoint is a narrow tool, and often the wrong one.

• Your agent runs on LangGraph, Pydantic AI, or the OpenAI Agents SDK. Use their native mechanisms (interrupt(), requires_approval=True, needs_approval=True). They pause the whole agent loop and resume it with the human's answer, which composes better inside those frameworks than an external queue does.
• You want Slack/email approval UX out of the box. That is a product, not a library. HumanLayer sells exactly it. holdpoint gives you hooks and a queue; the notification channel is your code.
• You need approvals across multiple services or machines. Use Temporal. Its signals + durable workflows are strictly stronger than a SQLite file, at the cost of running a cluster.
• The action is cheap to undo. A soft-deleted row does not need a human gate. Add approval only where mistakes are expensive and irreversible.

holdpoint fits when you have a Python process, a handful of scary function calls, and no appetite for new infrastructure.

Features

Guard any function. @hold.guard(kind=...) works on def and async def, with positional-only params, *args, **kwargs, and defaults (captured at submit time, so the reviewer sees the complete call). Full type inference: the IDE knows send_email(...) returns Action[str].

Patch before running. The reviewer is not a yes/no button. approve(id, actor="sava", patch={"subject": "Shorter"}) edits arguments before execution; unknown keys and unserializable values are rejected with exact messages, and the audit records both the original payload and the patch.

hold.approve(item.id, actor="sava", patch={"amount": 90})

Exactly-once execution. Approval is a compare-and-set in the store. Fifty concurrent approvals of one action execute it once; the other forty-nine wait and return the same stored result. See Guarantees for the precise claim.

Crash recovery. If the process dies between "approved" and "executed", hold.recover() at the next start finishes the job — or, if the crash happened mid-execution, surfaces the ambiguity instead of guessing.

report = hold.recover()          # call at startup, before serving traffic

TTLs with policies. ttl="24h" plus on_expire="expire" | "auto_approve" | "auto_reject". No background threads: call hold.expire_due() on your own cadence; overdue actions are un-approvable regardless.

Auto-approval rules. A predicate can wave routine actions through so the human only sees the interesting ones. Rules run the same machinery and leave the same audit trail (actor="rule:<name>").

def small_amounts(kind, payload):
    if kind == "wire" and payload["amount"] < 100:
        return "approve"
    return None

from holdpoint import MemoryStore

gated = HoldPoint(store=MemoryStore(), rules=[small_amounts])

Deduplication. dedup_key=lambda p: p["email"] collapses repeated proposals of the same action while one is live; the same key with a different payload is a loud conflict, never a silent overwrite.

Immutable audit. Who submitted, who decided, what was patched, what ran, what it returned or raised — append-only, enforced by database triggers, with secrets redacted via redact=("password",).

Hooks. @hold.on_pending (and on_approved / on_rejected / on_expired) is where you wire Telegram, Slack, or anything else. Notifications are best-effort by design; the queue is the source of truth.

Decide here, run there. By default the process that approves also executes. When your review surface should not hold the agent's credentials, approve without running and let a worker do it:

queued = send_email(to="ops@corp.com", subject="Weekly report")
hold.approve(queued.id, actor="sava", execute=False)     # the bot decides
print([a.status.value for a in hold.drain()])            # ['executed'] — the worker ran it

Both halves share one queue, one lease, one audit trail. Several workers can drain safely — an execution lease decides who runs what, and records who holds it.

Idempotency keys and verified recovery. current_action_id() inside a guarded function is a stable key for the downstream API, and guard(verify=...) lets recovery ask whether an interrupted side effect actually happened instead of paging a human:

from holdpoint import current_action_id


class Payments:  # stands in for your payment provider
    def __init__(self) -> None:
        self.seen: set[str] = set()

    def charge(self, amount: int, idempotency_key: str) -> str:
        self.seen.add(idempotency_key)
        return f"charged {amount}"

    def exists(self, key: str) -> bool:
        return key in self.seen


payments = Payments()


@hold.guard(kind="charge", verify=lambda payload, action_id: payments.exists(action_id))
def charge(amount: int) -> str:
    # the key is stable across retries, so the provider deduplicates for you
    return payments.charge(amount, idempotency_key=current_action_id() or "")


bill = charge(amount=100)
print(hold.approve(bill.id, actor="sava").value)  # "charged 100"

Retention. hold.purge(before=...) deletes settled actions with their audit entries, atomically; live actions are refused. Individual entries stay un-editable and un-deletable while their action exists.

from datetime import datetime, timedelta, timezone

cutoff = datetime.now(timezone.utc) - timedelta(days=90)
print(hold.purge(before=cutoff))  # 0 — nothing that old here yet

Secrets at rest. redact= hides values in the audit trail, redact_result= keeps a secret return value out of storage, and payload_codec= is the seam where you plug in encryption — holdpoint ships no cipher, because it ships no dependencies.

Know where things run. Approving executes the function unless you asked otherwise; hooks run synchronously in the thread that caused the event. Details in docs/concepts.md.

Async. await send_email(...), await hold.approve_async(...). Sync and async guards share one queue and one semantics.

Guarantees

What is promised, precisely:

• At most one execution at a time, ever. Concurrent approvals — threads, event loops, separate processes on one machine — collapse to one execution via an atomic claim; the rest share its result.
• Exactly one execution if the process survives the call. The normal case.
• After a crash: an approved action whose execution never started is completed by recover() safely. An action that died mid-execution is ambiguous by nature — if its guard has a verify predicate, recovery asks it and resolves the action for real; otherwise the default marks it failed with an explicit "side effect unknown" error for a human, and recover(interrupted="retry") chooses at-least-once, knowingly.
• One executor at a time, enforced by a lease. Workers and recovery never take an action another executor still owns; recover() is safe to run alongside live traffic.
• Every state change commits atomically with its audit entry. A crash cannot separate what happened from the record of it.

What is not promised: coordination across machines (single shared SQLite file, one host), guaranteed hook delivery (best-effort; poll the queue for truth), retries of failed actions (a failed action stays failed until a human re-submits), and protection from os._exit() between a side effect and its recording — that gap is physics, and holdpoint chooses to expose it rather than pretend.

Compared honestly

The bold cells in the last rows are where holdpoint loses. If those rows are your requirements, take the other column's tool.

Status

v0.1.x. The public API (everything importable from holdpoint) follows semver: breaking changes only with a major-version bump; 0.x minor bumps may extend but not break. The SQLite schema is versioned and migrates forward automatically; downgrade is refused explicitly.

License

FSL-1.1-ALv2 (Functional Source License): free for any use except competing commercial hosting, and each release becomes Apache-2.0 two years after publication. Contributions require the short CLA.