Backfill a year of history

​

The scenario

​

Step 1 — Queue a single write

queued = client.record(
    space_id=space,
    content="The incident review moved to Wednesdays.",
    background=True,
)
print("queued as", queued["job_id"])

​

Step 2 — Send them in batches

batch = client.record_batch(
    space_id=space,
    items=[
        {"content": "Sprint 41 closed with 34 points."},
        {"content": "Sprint 42 planning is Thursday.", "user_id": "scrum-bot"},
    ],
)
print("accepted:", batch["accepted"])

​

Step 3 — Poll for the outcome

def wait(job_id):
    while True:
        job = client.get_job(space_id=space, job_id=job_id)
        if job["status"] in ("completed", "failed"):
            return job
        time.sleep(5)

done = wait(batch["job_id"])
print("stored", done["memory_count"], "memories")

​

Evals

1. Queue one write, poll to completion, assert memory_ids is non-empty.
2. Time a blocking record against a background one. The difference is why this exists.
3. Import a hundred items and check memory_count against what you sent — extraction can produce more memories than items, and that is expected.

​

Guardrails

• Throughput is bounded platform-side. A large import takes real time; size your expectations from a measured batch rather than from the accept latency.
• accepted is not stored. It means the queue took them. The job is what tells you they landed.
• A failed job needs handling. Poll for failed as well as completed, or a silent import gap looks like a retrieval problem weeks later.

​

The script