AI Agent State Between Runs: Context, Keys and Audit

Keep the facts in a key-value store that belongs to the agent process, inject a snapshot of it into the prompt at the start of every run, and checkpoint it to durable storage after each run. In Promptise that store is AgentContext. You seed it with initial_state (or world: in a manifest), restrict writes with writable_keys, and every put() records an audit entry with a timestamp and a source. The runtime adds the current state to each run's prompt, and with a journal it saves the state after every run so you can restore it in a new process.

context=ContextConfig(
    initial_state={"paused_pipelines": ["staging-etl"], "open_incidents": {}},
    writable_keys=["paused_pipelines", "open_incidents"],
)
process.context.put("open_incidents", {"billing": "INC-311"}, source="ops-anna")

• State is exact and always in the prompt. Memory is searched, and conversation history is trimmed. State is neither.

• `writable_keys` is an allow-list. A put() on any other key raises KeyError, and the API returns 403.

• Every write is audited. state_history(key) returns each value with its time and source.

• The agent can't write state by itself. Give it a tool that does, or write state from your own code.

• A restart of the Python process starts from `initial_state` unless you restore the journal's checkpoint.

An autonomous agent can carry three kinds of information between runs, and each answers a different question:

State is the agent's blackboard: a small, structured record that several writers share, each write attributed. The conversation buffer is short-term memory. It is useful when a run needs to know what the last run said, but a fact that matters must not depend on staying inside a window of 100 messages. Long-term memory is for recall: the AI agent memory guide covers it in depth.

How state moves through the runs of one process:

start: state = initial_state (or the restored checkpoint)
   │
   ▼
run 1 ── prompt gets "[Context State] {...}" ── agent replies
   │                                        └── journal checkpoint: context_state
   ▼
your code / PATCH /context / a tool ── put(key, value, source) ── audit entry
   │                                     (writable_keys checked)
   ▼
run 2 ── prompt gets the new snapshot ── agent replies ── checkpoint
   │
   ▼
process dies ── new process: ReplayEngine.recover() ── seed initial_state

Use context state when a run must know a fact for certain: a pipeline is paused, an incident is open, item 4,212 is the last one processed. Use it when several parties need to change that fact, such as an operator, a webhook handler and the agent itself, and you need to know who changed it.

Don't use it for large data. The whole snapshot goes into every prompt, so a list of 5,000 IDs costs tokens on every run. Store large data in a database and keep a pointer or a summary in state. Don't use it for secrets either: state is printed into the prompt, returned by GET /context and written to the journal. Use the process's secret scope instead. And don't use it for things the agent should merely recall, such as how a similar incident was solved last month. That is memory.

• Python 3.10 or later and Promptise Foundry 1.4.3 or later.

• An OpenAI API key in OPENAI_API_KEY. Steps 3 and 4 make three openai:gpt-5-mini runs in total, a few cents at most.

• About 20 minutes.

pip install "promptise>=1.4.3"

AgentContext works without a process, which makes its rules easy to see:

"""AgentContext on its own: state, writable keys, audit trail, env and files."""
import os

from promptise.runtime import AgentContext

os.environ["AGENT_REGION"] = "eu-west-1"
os.environ["DATABASE_PASSWORD"] = "not-for-the-agent"

ctx = AgentContext(
    initial_state={"last_report_id": 0, "paused_pipelines": [], "on_call": "ops-anna"},
    writable_keys=["last_report_id", "paused_pipelines"],
    env_prefix="AGENT_",
    file_mounts={"reports": "/srv/reports"},
)

ctx.put("last_report_id", 41, source="trigger")
ctx.put("paused_pipelines", ["staging-etl"], source="ops")
ctx.put("last_report_id", 42, source="trigger")

try:
    ctx.put("on_call", "ops-ben", source="agent")
except KeyError as exc:
    print("refused:", exc)

print("snapshot:", ctx.state_snapshot())
print("missing key:", ctx.get("error_budget", "unknown"))
for entry in ctx.state_history("last_report_id"):
    print(f"  last_report_id={entry.value!r:<4} source={entry.source}")

print("env:", ctx.env)
print("files:", ctx.files)

# get() returns the live object: changing it in place skips the audit trail.
ctx.get("paused_pipelines").append("search-index")
print("after in-place append:", ctx.get("paused_pipelines"),
      "| history entries:", len(ctx.state_history("paused_pipelines")))

refused: "Key 'on_call' is not writable. Allowed: ['last_report_id', 'paused_pipelines']"
snapshot: {'last_report_id': 42, 'paused_pipelines': ['staging-etl'], 'on_call': 'ops-anna'}
missing key: unknown
  last_report_id=0    source=system
  last_report_id=41   source=trigger
  last_report_id=42   source=trigger
env: {'AGENT_REGION': 'eu-west-1'}
files: {'reports': '/srv/reports'}
after in-place append: ['staging-etl', 'search-index'] | history entries: 2

What each part does:

• `initial_state` fills the store and records each key with source system.

• `put(key, value, source="agent")` checks writable_keys, stores the value and appends a StateEntry(key, value, timestamp, source). The source is free text, so name the writer: ops-anna, trigger, api.

• `get(key, default)` reads a value. state_snapshot() returns a shallow copy of the whole store, and state_keys() lists the keys.

• `state_history(key)` returns the audit trail, oldest first. History entries are deep copies, so a later change can't rewrite them.

• `env` returns the environment variables that start with env_prefix (default AGENT_), with the prefix kept. DATABASE_PASSWORD is not in the output.

• `files` returns the file_mounts mapping of names to paths. The context does no file I/O itself.

The last line shows a trap. get() returns the live object, so appending to the list changed the state without a put(), and the audit trail still has two entries. Treat values as immutable: build a new list and put() it.

writable_keys=[], the default, makes every key writable. The check applies to every put(), including your own code's. Only initial_state and a restore with AgentContext.from_dict() bypass it. env and files are for your code and tools: the runtime doesn't add them to the prompt.

In a process, ContextConfig holds the same settings plus memory and conversation options:

{}

{}

In a .agent manifest, the initial state is called world:, and the other context settings go under context: and memory::

version: "1.0"
name: pipeline-watch
model: openai:gpt-5-mini
instructions: >
  You read the nightly pipeline report. Skip pipelines listed in paused_pipelines.
triggers:
  - type: cron
    cron_expression: "0 6 * * *"
world:
  paused_pipelines: [staging-etl]
  open_incidents: {}
context:
  writable_keys: [paused_pipelines, open_incidents]
  env_prefix: PIPELINE_
  file_mounts:
    reports: /srv/reports
memory:
  provider: chroma
  persist_directory: .promptise/memory
  collection: pipeline-watch
  auto_store: true
journal:
  level: checkpoint

"""Load the same state settings from a .agent manifest."""
from promptise.runtime.manifest import load_manifest, manifest_to_process_config

config = manifest_to_process_config(load_manifest("pipeline-watch.agent"))
ctx = config.context
print("initial_state:", ctx.initial_state)
print("writable_keys:", ctx.writable_keys)
print("env_prefix:", ctx.env_prefix, "| file_mounts:", ctx.file_mounts)
print("memory:", ctx.memory_provider, ctx.memory_persist_directory, ctx.memory_collection,
      "auto_store =", ctx.memory_auto_store)
print("journal:", config.journal.level)

initial_state: {'paused_pipelines': ['staging-etl'], 'open_incidents': {}}
writable_keys: ['paused_pipelines', 'open_incidents']
env_prefix: PIPELINE_ | file_mounts: {'reports': '/srv/reports'}
memory: chroma .promptise/memory pipeline-watch auto_store = True
journal: checkpoint

Two manifest details: the context: block reads only writable_keys, file_mounts, env_prefix and conversation_max_messages, so an initial_state there is ignored. And a manifest's journal defaults to level: checkpoint, while a ProcessConfig in Python has no journal until you add one.

state_run.py runs the same nightly report twice. Between the runs, an operator records an open incident in state. Conversation history is off, so state is the only thing that carries over:

"""An agent that reads its state on every run, while your code changes it in between."""
import asyncio

from promptise.runtime import AgentProcess, ContextConfig, JournalConfig, ProcessConfig
from promptise.runtime.journal import FileJournal
from promptise.runtime.triggers.base import TriggerEvent

config = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions=(
        "You read the nightly pipeline report. List each failed pipeline in one line. "
        "Skip pipelines listed in paused_pipelines. If a failed pipeline has an entry "
        "in open_incidents, add the incident note instead of raising a new alarm."
    ),
    context=ContextConfig(
        initial_state={"paused_pipelines": ["staging-etl"], "open_incidents": {}},
        writable_keys=["paused_pipelines", "open_incidents"],
        conversation_history=False,  # runs share state, not chat history
        memory_provider="in_memory",
        memory_auto_store=True,
    ),
    journal=JournalConfig(level="checkpoint", path=".promptise/journal"),
)
REPORT = {"report": "staging-etl: FAILED\nbilling: FAILED\norders: OK"}


async def run_once(process: AgentProcess, journal: FileJournal, n: int) -> None:
    await process.inject(TriggerEvent(trigger_id="nightly", trigger_type="cron", payload=REPORT))
    while process.status()["invocation_count"] < n:
        await asyncio.sleep(0.2)
    results = [e for e in await journal.read("pipeline-watch") if e.entry_type == "invocation_result"]
    print(f"run {n}: {results[-1].data['response']}")


async def main() -> None:
    process = AgentProcess("pipeline-watch", config)
    journal = FileJournal(".promptise/journal")
    await process.start()

    await run_once(process, journal, 1)

    # Between runs, your code (or an operator through the API) updates the state.
    process.context.put("open_incidents", {"billing": "INC-311, fix deploys at 14:00"},
                        source="ops-anna")
    await run_once(process, journal, 2)

    checkpoint = await journal.last_checkpoint("pipeline-watch")
    print("checkpointed state:", checkpoint["context_state"])
    hits = await process.context.search_memory("billing", limit=5)
    print("memories about billing:", len(hits))
    await process.stop()


if __name__ == "__main__":
    asyncio.run(main())

run 1: billing: FAILED — no open incident; raise a new alarm.
run 2: billing: FAILED — INC-311, fix deploys at 14:00
checkpointed state: {'paused_pipelines': ['staging-etl'], 'open_incidents': {'billing': 'INC-311, fix deploys at 14:00'}}
memories about billing: 2

Both runs skipped staging-etl because it is in paused_pipelines. In run 1 nothing was known about billing, so the agent raised an alarm. In run 2 it found the incident note in state and reported it instead. At the start of each run, the runtime added the snapshot to the prompt as a system message of the form [Context State] {...}. It adds nothing when the state is empty.

After each successful run, the journal checkpoint stored the snapshot, which the script read back. memory_auto_store=True also stored each run's input and reply in long-term memory, which is why the keyword search for "billing" found two entries. Memory recorded what happened. State recorded what is true now.

The agent reads state but has no built-in tool to write it, in strict or open mode. When the agent itself should record a fact, give it a tool. The cleanest route goes through the control API, because the API applies writable_keys and records the write with source api. This MCP server has one tool that adds a pipeline to paused_pipelines:

"""MCP tool that lets the agent change its own state through the control API."""
import os

import httpx
from promptise.mcp.server import MCPServer

URL = f"{os.environ['CONTROL_API']}/api/v1/processes/{os.environ['PROCESS_NAME']}/context"
HEADERS = {"Authorization": f"Bearer {os.environ['ORCHESTRATION_API_TOKEN']}"}
server = MCPServer("state-tools")


@server.tool()
async def pause_pipeline(pipeline: str) -> dict:
    """Stop reporting a pipeline, for example during a maintenance window."""
    async with httpx.AsyncClient(headers=HEADERS) as http:
        state = (await http.get(URL)).json()["state"]
        paused = sorted(set(state.get("paused_pipelines", [])) | {pipeline})
        resp = await http.patch(URL, json={"state": {"paused_pipelines": paused}})
    if resp.status_code != 200:
        return {"error": resp.json()["error"]["message"]}
    return {"paused_pipelines": paused}


if __name__ == "__main__":
    server.run(transport="stdio")

"""The agent pauses a pipeline itself; the change lands in its context with source 'api'."""
import asyncio
import os
import sys

from promptise.config import StdioServerSpec
from promptise.runtime import AgentRuntime, OrchestrationAPI
from promptise.runtime.journal import FileJournal
from promptise.runtime.triggers.base import TriggerEvent
from state_run import config

config = config.model_copy(deep=True)
config.instructions += (" If the report says a pipeline is in a maintenance window, "
                        "call pause_pipeline for it and don't list it.")
config.servers = {"state": StdioServerSpec(
    command=sys.executable, args=["state_tools.py"],
    env={"CONTROL_API": "http://127.0.0.1:9100", "PROCESS_NAME": "pipeline-watch",
         "ORCHESTRATION_API_TOKEN": os.environ["ORCHESTRATION_API_TOKEN"]})}


process_journal = FileJournal(".promptise/journal")


async def main() -> None:
    runtime = AgentRuntime()
    process = await runtime.add_process("pipeline-watch", config)
    api = OrchestrationAPI(runtime, port=9100, auth_token="${ORCHESTRATION_API_TOKEN}")
    await api.start()
    await process.start()

    await process.inject(TriggerEvent(trigger_id="nightly", trigger_type="cron", payload={
        "report": "orders: FAILED (maintenance window until Friday)\nbilling: OK"}))
    while process.status()["invocation_count"] < 1:
        await asyncio.sleep(0.2)

    results = [e for e in await process_journal.read("pipeline-watch")
               if e.entry_type == "invocation_result"]
    print("reply:", results[-1].data["response"])
    for entry in process.context.state_history("paused_pipelines"):
        print(f"paused_pipelines={entry.value} source={entry.source}")
    await process.stop()
    await api.stop()


if __name__ == "__main__":
    asyncio.run(main())

reply: No failed pipelines to list — "orders" was in a maintenance window and has been paused.
paused_pipelines=['staging-etl'] source=system
paused_pipelines=['orders', 'staging-etl'] source=api

The report said orders was in a maintenance window, so the agent called pause_pipeline("orders"), and the audit trail shows the write with source api. The tool sends the full new list, because PATCH /context replaces a key's value. Give the agent tools that change one specific key, rather than a generic "set any key" tool, so the tool's name and description say what the agent may change. The token in the tool's environment can do everything the API allows, so keep the server on the same host. The control API guide covers its security.

Operators and other services use the same two routes:

TOKEN=demo-token-change-me
API=http://127.0.0.1:9100/api/v1/processes/pipeline-watch

curl -s $API/context -H "Authorization: Bearer $TOKEN"; echo

curl -s -X PATCH $API/context -H "Authorization: Bearer $TOKEN" \
  -d '{"state": {"open_incidents": {"billing": "INC-311, fix deploys at 14:00"}}}'; echo

# One allowed key and one that is not writable:
curl -s -X PATCH $API/context -H "Authorization: Bearer $TOKEN" \
  -d '{"state": {"paused_pipelines": ["staging-etl", "orders"], "owner": "ops-ben"}}'; echo

curl -s $API/context -H "Authorization: Bearer $TOKEN"; echo

{"state": {"paused_pipelines": ["staging-etl"], "open_incidents": {}}, "writable_keys": ["paused_pipelines", "open_incidents"]}
{"name": "pipeline-watch", "updated_keys": ["open_incidents"]}
{"error": {"code": "WRITE_DENIED", "message": "Cannot write key 'owner': \"Key 'owner' is not writable. Allowed: ['open_incidents', 'paused_pipelines']\""}}
{"state": {"paused_pipelines": ["staging-etl", "orders"], "open_incidents": {"billing": "INC-311, fix deploys at 14:00"}}, "writable_keys": ["paused_pipelines", "open_incidents"]}

GET /context returns the state and the writable keys. PATCH /context takes {"state": {key: value}} and writes each key with source api. Read the third call closely: the patch held one allowed key and one denied key. The API answered 403 WRITE_DENIED, yet the final read shows that paused_pipelines was changed. Keys are written one at a time and the first denied key stops the loop, so the keys before it are already written. Send one key per request when you need all-or-nothing behavior.

A write over the API changes the state in memory. It reaches the journal with the next run's checkpoint, so if the process dies before its next run, the change is lost. The demo's journal folder held no checkpoint after these calls, because no run had happened.

Within one Python process, state survives a stop, a restart and an instruction change, because they keep the same AgentContext. A new Python process, after a deploy or a crash, builds a new context from initial_state. To continue where the old one stopped, read the last checkpoint with ReplayEngine and use it as the initial state:

"""A new Python process: start fresh, or seed the state from the journal's checkpoint."""
import asyncio

from promptise.runtime import AgentProcess
from promptise.runtime.journal import FileJournal, ReplayEngine
from state_run import config


async def main() -> None:
    fresh = AgentProcess("pipeline-watch", config)
    print("without recovery:", fresh.context.state_snapshot())

    recovered = await ReplayEngine(FileJournal(".promptise/journal")).recover("pipeline-watch")
    seeded = config.model_copy(deep=True)
    seeded.context.initial_state = recovered["context_state"]
    process = AgentProcess("pipeline-watch", seeded)
    print("with recovery:   ", process.context.state_snapshot())
    print("history:", [(e.value, e.source) for e in process.context.state_history("open_incidents")])


if __name__ == "__main__":
    asyncio.run(main())

without recovery: {'paused_pipelines': ['staging-etl'], 'open_incidents': {}}
with recovery:    {'paused_pipelines': ['staging-etl'], 'open_incidents': {'billing': 'INC-311, fix deploys at 14:00'}}
history: [({'billing': 'INC-311, fix deploys at 14:00'}, 'system')]

Without recovery, the open incident is gone. With it, the state matches the end of the last run. The audit trail does not come back: the checkpoint holds the snapshot, so recovered keys start with one system entry. Keep the audit trail somewhere durable if you need it across deploys. recover() also replays any context_update journal entries written after the checkpoint, which lets you persist a write before the next run. The durable AI agent guide shows that pattern and how to make side effects safe to repeat.

state-demo/
├── context_basics.py       # AgentContext rules (step 1)
├── pipeline-watch.agent    # the same settings as a manifest (step 2)
├── load_manifest.py
├── state_run.py            # two runs, state changed in between (step 3)
├── state_tools.py          # MCP tool that writes state through the API (step 4)
├── agent_writes_state.py
├── serve_state.py          # the process behind the control API (step 5)
├── state_api.sh
└── recover_state.py        # seed a new process from the checkpoint (step 6)

pip install "promptise>=1.4.3"
export OPENAI_API_KEY=sk-... ORCHESTRATION_API_TOKEN=demo-token-change-me
python context_basics.py && python load_manifest.py
python state_run.py && python recover_state.py
python agent_writes_state.py
python serve_state.py &   # then: bash state_api.sh

serve_state.py puts the process from state_run.py behind the API:

"""Serve the pipeline-watch process behind the control API."""
import asyncio

from promptise.runtime import AgentRuntime, OrchestrationAPI
from state_run import config


async def main() -> None:
    runtime = AgentRuntime()
    await runtime.add_process("pipeline-watch", config)
    await runtime.start_all()
    api = OrchestrationAPI(runtime, port=9100, auth_token="${ORCHESTRATION_API_TOKEN}")
    await api.start()
    try:
        await asyncio.Event().wait()
    finally:
        await api.stop()
        await runtime.stop_all()


if __name__ == "__main__":
    asyncio.run(main())

Give each writer the narrowest keys

writable_keys is one list per process, shared by every writer. Keep keys that only your system should set, such as an owner or a tenant ID, in initial_state and out of writable_keys, so nothing can change them at runtime. Give the agent tools that each change one key. Use the source argument to name every writer, because the audit trail is only as useful as its sources.

Keep state small and stable

The snapshot is printed into every run's prompt as a Python dict. Every key costs tokens on every run, and a value that changes often makes prompts harder to cache. Keep values short: IDs, flags, short notes. Put lists that grow into a database, and expose them through tools.

Turn on the journal and recover on start

Add JournalConfig(level="checkpoint") so state is saved after every run, and run the step 6 recovery when your service starts. The checkpoint is written only after a successful run, so state written between runs is at risk until the next one. If that matters, append a context_update entry yourself when you write. Back up the journal folder with the rest of your data.

Choose a memory provider that persists

in_memory is a keyword index that lives in the Python process, for tests only. For memory that survives deploys, use chroma with a memory_persist_directory, or mem0. In 1.4.3 a process stop also closes its memory provider, and a restart in the same Python process doesn't open it again: in_memory comes back empty and Chroma raises "ChromaProvider is closed". After /restart, long-term memory works again only in a new Python process.

Watch for concurrent writes

put() replaces a value. Two writers that read a list, add an item and write it back can lose an item. Within one agent with concurrency=1 this is rare. With several writers, keep one writer per key, or serialize updates in your code.

• The agent has no built-in state tool. It reads state; writing needs your code, the API or a tool you build.

• The runtime doesn't restore state on start. Recovery is a few lines of your code.

• The audit trail lives in memory. It is lost with the Python process, and the journal stores snapshots only.

• `writable_keys` has no per-writer rules. The agent's tools, the API and your code share one allow-list.

• No transactions. API patches can apply partly, and concurrent writers can overwrite each other.

• Values should be JSON types. The file journal writes anything else as its string form, so a datetime or a set comes back as a string after recovery.

How do AI agents remember things between runs?

Through three mechanisms: explicit state, conversation history and long-term memory. State holds exact named facts and goes into every prompt. Conversation history carries the last runs' messages. Long-term memory is searched for notes related to the current input. For facts a run must get right, such as which pipelines are paused, use state. In Promptise that is AgentContext, saved by the journal after each run.

What is the difference between agent state and agent memory?

State is a small key-value store whose whole content is in every prompt, with exact values and an audit trail. Memory is a searchable store of text, and each run sees only the entries that match its input best. Use state for facts you need to trust, like an open incident ID, and memory for recall, like how a similar incident was handled.

Does agent state survive a restart?

Within one Python process, yes: stopping and starting a process keeps its context. A new Python process starts from initial_state again. To continue, enable the journal with level="checkpoint", read the last checkpoint with ReplayEngine(journal).recover(name) on start-up, and pass its context_state as the new initial state.

How do I stop an agent from overwriting important state?

List the keys it may change in writable_keys. Every other key raises KeyError on put() and returns 403 WRITE_DENIED over the API. Keep system-owned values in initial_state and out of that list. Give the agent narrow tools that each change one key, and record a meaningful source with every write so you can see who changed what.

Can I change an agent's state while it is running?

Yes. In Python, call process.context.put(key, value, source=...). Over HTTP, send PATCH /api/v1/processes/{name}/context with {"state": {...}}. The next run sees the new value in its prompt. A run that is already in progress keeps the snapshot it started with. The change reaches the journal with the next checkpoint.

Should I put API keys or passwords in agent state?

No. State is printed into the prompt, returned by GET /context and stored in the journal, so a secret there would leak to the model provider, to API users and to disk. Use the process's secret scope, which keeps values encrypted in memory, never returns them over the API, and lets you rotate and revoke them.

• Make recovery and side effects crash-safe: durable AI agents.

• Add long-term memory and choose a provider: AI agent memory.

• Read and change state over HTTP: AI agent control API.

• Tell a running agent something temporary: message a running AI agent.

• Design what goes into each prompt: context engineering for AI agents.

• Reference: agent context, conversation buffer and journal replay in the Promptise docs.