Messages, Roles & Conversation State

# script_id: day_008_messages_and_state/stateless_demo
# Two independent calls forget each other; one shared list fixes it. ≈ $0.001.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")  # built on Day 2's atlas/config.py
client = OpenAI()

FACT = ("Orbit docs: ORB-1101 LeaseExpired means a job outlived its lease and may run twice. "
        "Fix: let heartbeats renew the lease ([worker] heartbeat_seconds) or raise "
        "[worker] lease_seconds for long jobs.")
turn1 = f"{FACT}\n\nMy worker logs ORB-1101. What is it?"
turn2 = "How do I fix it?"


def reply(input_items: str | list) -> tuple[str, list]:
    resp = client.responses.create(model=MODEL, input=input_items, reasoning={"effort": "low"})
    return resp.output_text, resp.output


# A) Two independent calls: the second request contains only turn2.
reply(turn1)
alone, _ = reply(turn2)
print("WITHOUT history:", alone[:150])

# B) Carry the conversation yourself: turn 1, the model's output, then turn 2.
history: list = [{"role": "user", "content": turn1}]
_, items = reply(history)
history += items                      # the model's whole turn, not only its text
history.append({"role": "user", "content": turn2})
together, _ = reply(history)
print("WITH history:   ", together[:150])

WITHOUT history: I'm not sure what "it" refers to. Could you tell me which error or tool you're working with?
WITH history:    Make sure heartbeats can renew the lease: set [worker] heartbeat_seconds well below lease_seconds, or raise [worker] lease_seconds for long jobs.
(illustrative; your wording will differ)

# script_id: day_008_messages_and_state/turn2_by_hand
# The exact request body for turn 2 of a chat. Runs offline: no API call, no key.
import json

instructions = "You are Atlas, assistant to Orbit's maintainers. Answer in 2 sentences."
turn_2_input = [
    {"role": "user", "content": "My worker logs ORB-1101. What is it?"},
    {"role": "assistant", "content": "ORB-1101 LeaseExpired: a job outlived its lease, so it may run twice."},
    {"role": "user", "content": "How do I fix it?"},
]
print("instructions:", instructions)
print(json.dumps(turn_2_input, indent=2))
print(f"{len(turn_2_input)} messages; the model reads all of them on every call.")

instructions: You are Atlas, assistant to Orbit's maintainers. Answer in 2 sentences.
[
  {
    "role": "user",
    "content": "My worker logs ORB-1101. What is it?"
  },
  {
    "role": "assistant",
    "content": "ORB-1101 LeaseExpired: a job outlived its lease, so it may run twice."
  },
  {
    "role": "user",
    "content": "How do I fix it?"
  }
]
3 messages; the model reads all of them on every call.

# script_id: day_008_messages_and_state/roles_placement
# instructions vs developer vs user, in one call. ≈ $0.001.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

INSTRUCTIONS = "You are Atlas, assistant to Orbit's maintainers. Answer in at most 3 sentences."
input_items = [
    # Written by your app (e.g. from the account DB), not typed by the user.
    {"role": "developer", "content": (
        "This user runs Orbit 2.9.0. Facts: on 2.x the lease length is the ORBIT_LEASE "
        "environment variable; on 3.x it is [worker] lease_seconds in orbit.toml.")},
    # The end user's words, sent as-is and never merged into instructions.
    {"role": "user", "content": "Where do I set the lease length?"},
]
resp = client.responses.create(model=MODEL, instructions=INSTRUCTIONS, input=input_items,
                               reasoning={"effort": "low"})
print(resp.output_text)

Since you're on Orbit 2.9.0, set the lease length with the ORBIT_LEASE environment variable. On 3.x it moves to [worker] lease_seconds in orbit.toml.
(illustrative; your wording will differ)

# script_id: day_008_messages_and_state/output_items
# Inspect output items, phase, output_text and usage. ≈ $0.001.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY, ORBIT_DATA (path to datasets/orbit)
import os
from pathlib import Path

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
client = OpenAI()

faq = (DATA_DIR / "docs" / "faq.md").read_text(encoding="utf-8")[:2400]  # a slice keeps it cheap
resp = client.responses.create(
    model=MODEL,
    instructions=f"Answer only from these Orbit docs.\n<facts>\n{faq}\n</facts>",
    input="Does Orbit run each job exactly once?",
    reasoning={"effort": "low"},
)
print("item types:", [item.type for item in resp.output])
for item in resp.output:
    if item.type == "message":                  # only message items carry visible text
        print("phase:", item.phase, "| parts:", [part.type for part in item.content])
print("output_text:", resp.output_text)
u = resp.usage
print(f"usage: input={u.input_tokens} output={u.output_tokens} "
      f"(reasoning={u.output_tokens_details.reasoning_tokens})")

item types: ['reasoning', 'message']
phase: final_answer | parts: ['output_text']
output_text: No. Orbit delivers jobs at least once: a job whose lease expires (ORB-1101) or whose worker dies can run again, so make side effects idempotent.
usage: input=631 output=88 (reasoning=41)
(illustrative; your items and numbers may differ)

# script_id: day_008_messages_and_state/minimal_chat
# The smallest correct chat loop. Type /quit or an empty line to stop. ≈ $0.001 per turn.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
INSTRUCTIONS = "You are Atlas, a concise assistant for Orbit's maintainers."
history: list = []                       # the conversation lives here, in your process

while (user_text := input("you> ").strip()) not in ("", "/quit"):
    history.append({"role": "user", "content": user_text})
    resp = client.responses.create(model=MODEL, instructions=INSTRUCTIONS, input=history,
                                   reasoning={"effort": "low"})  # default is medium: be explicit
    history += resp.output               # every item: reasoning + message (with its phase)
    print("atlas>", resp.output_text)

you> My name is Priya and I maintain the Redis broker.
atlas> Nice to meet you, Priya. What's going on with the Redis broker?
you> What's my name, and what do I maintain?
atlas> You're Priya, and you maintain Orbit's Redis broker.
(illustrative)

# script_id: day_008_messages_and_state/token_growth
# Five scripted turns; watch input tokens grow. Run: python token_growth.py [items|text|none]
# ≈ $0.003 per run. Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY
import os
import sys

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
MODE = sys.argv[1] if len(sys.argv) > 1 else "items"
QUESTIONS = ["My Orbit worker logs ORB-1101 LeaseExpired. In one sentence, what might that mean?",
             "Name one setting you'd check first.", "What goes wrong if I set it far too high?",
             "Give me a one-line summary of the fix.", "Which setting did you tell me to check first?"]

history: list = []
billed = 0
for turn, question in enumerate(QUESTIONS, start=1):
    history.append({"role": "user", "content": question})
    resp = client.responses.create(model=MODEL, input=history, reasoning={"effort": "low"})
    if MODE == "items":
        history += resp.output                     # correct: everything the model produced
    elif MODE == "text":                           # works for plain text, drops reasoning + phase
        history.append({"role": "assistant", "content": resp.output_text})
    # MODE == "none": the bug, only user turns are kept
    billed += resp.usage.input_tokens
    print(f"turn {turn} · input {resp.usage.input_tokens:>4} tok · billed so far {billed:>5} · "
          f"{resp.output_text[:45]!r}")

$ python token_growth.py items
turn 1 · input   32 tok · billed so far    32 · 'ORB-1101 LeaseExpired usually means a job ran'
turn 2 · input   95 tok · billed so far   127 · 'Check the lease length (lease_seconds) first.'
turn 3 · input  139 tok · billed so far   266 · 'Too high a lease delays recovery: if a worker'
turn 4 · input  201 tok · billed so far   467 · 'Keep heartbeats well inside the lease and siz'
turn 5 · input  248 tok · billed so far   715 · 'I suggested checking the lease length (lease_'
$ python token_growth.py none
turn 5 · input   78 tok · billed so far   250 · "I don't see an earlier suggestion from me in"
(illustrative; your numbers will differ)

# script_id: day_008_messages_and_state/server_state
# Same 3-turn chat two ways; compare input tokens. ≈ $0.002.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
INSTRUCTIONS = "You are Atlas, a concise assistant for Orbit's maintainers."
QUESTIONS = ["What does a job lease protect against?", "And what renews it?", "Summarize in one line."]

manual: list = []
prev_id: str | None = None
print("turn | manual input tok | server input tok")
for turn, question in enumerate(QUESTIONS, start=1):
    manual.append({"role": "user", "content": question})
    a = client.responses.create(model=MODEL, instructions=INSTRUCTIONS, input=manual,
                                reasoning={"effort": "low"})
    manual += a.output
    b = client.responses.create(model=MODEL, instructions=INSTRUCTIONS,  # not carried over: re-send
                                input=question, previous_response_id=prev_id,
                                reasoning={"effort": "low"})
    prev_id = b.id                       # chaining needs a stored response (store defaults to True)
    print(f"{turn:>4} | {a.usage.input_tokens:>16} | {b.usage.input_tokens:>16}")

turn | manual input tok | server input tok
   1 |               24 |               24
   2 |               71 |               69
   3 |              113 |              116
(illustrative; the two columns track each other)

# script_id: day_008_messages_and_state/chat_cli
# Atlas CLI chat: persona + Orbit fact block + explicit history. ≈ $0.01 for a 10-turn chat.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY, ORBIT_DATA (ATLAS_MODEL optional)
# Run: python chat_cli.py   (commands: /reset /tokens /history /quit)
import os
from pathlib import Path

import openai
from openai import OpenAI
from openai.types.responses import Response

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")  # built on Day 2's atlas/config.py
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
EFFORT = "low"
client = OpenAI()

# Anatomy: persona, goal, rules, format, escalation. The facts are appended below.
ATLAS_PERSONA = """You are Atlas, an assistant for the maintainers of Orbit, an open-source Python background-jobs library.
Goal: help maintainers answer user questions quickly and correctly.
Rules:
- Use only the facts between the <facts> tags. If the answer is not there, say you don't know and suggest which doc page to check.
- Never invent config keys, versions or error codes.
Format: at most 4 sentences or 4 bullets. Put commands and config keys in backticks.
Escalation: if a user reports data loss or a security problem, say a human maintainer must look at it."""


def load_facts(path: Path, max_chars: int = 2400) -> str:
    """Read a docs page, drop its YAML front matter, and cut at a paragraph boundary."""
    text = path.read_text(encoding="utf-8")
    if text.startswith("---"):
        text = text.split("---", 2)[2]          # ["", front matter, body]
    text = text.strip()
    if len(text) <= max_chars:
        return text
    cut = text.rfind("\n\n", 0, max_chars)      # don't end mid-sentence
    return text[: cut if cut > 0 else max_chars]

# script_id: day_008_messages_and_state/chat_cli
INSTRUCTIONS = f"{ATLAS_PERSONA}\n\n<facts>\n{load_facts(DATA_DIR / 'docs' / 'faq.md')}\n</facts>"


def chat_turn(history: list, user_text: str, *, instructions: str) -> tuple[list, Response]:
    """Send one user turn with the full history; return (new history, response)."""
    turn_input = [*history, {"role": "user", "content": user_text}]
    resp = client.responses.create(model=MODEL, instructions=instructions,  # every turn
                                   input=turn_input, reasoning={"effort": EFFORT})
    return [*turn_input, *resp.output], resp


def describe(item) -> str:
    """One line per history entry: dicts are your inputs, objects are model output items."""
    if isinstance(item, dict):
        return f"{item['role']:<9} {item['content'][:60]!r}"
    if item.type == "message":
        text = "".join(part.text for part in item.content if part.type == "output_text")
        return f"{'assistant':<9} {text[:60]!r} (phase={item.phase})"
    return f"[{item.type} item]"

# script_id: day_008_messages_and_state/chat_cli
def main() -> None:
    history: list = []
    last_usage = None
    turn = 0
    print(f"Atlas on {MODEL}. Commands: /reset /tokens /history /quit")
    while (user_text := input("you> ").strip()) not in ("", "/quit"):
        if user_text == "/reset":
            history, turn = [], 0
            print("(history cleared: the model now knows nothing about this chat)")
        elif user_text == "/tokens":
            if last_usage:
                print(f"last turn: input {last_usage.input_tokens} · output {last_usage.output_tokens} tok")
            else:
                print("(no turns yet)")
        elif user_text == "/history":
            for i, item in enumerate(history):
                print(f"{i:>3} {describe(item)}")
        else:
            try:
                history, resp = chat_turn(history, user_text, instructions=INSTRUCTIONS)
            except openai.APIError as err:      # history is unchanged, so the user can retry
                print(f"(API error: {type(err).__name__}; nothing was added to history)")
                continue
            turn, last_usage = turn + 1, resp.usage
            print(f"atlas> {resp.output_text}\n[turn {turn} · input {last_usage.input_tokens} tok]")


if __name__ == "__main__":
    main()

Atlas on gpt-6-luna. Commands: /reset /tokens /history /quit
you> My worker logs ORB-1101. Can a job run twice?
atlas> Yes. Orbit delivers jobs at least once, and a job whose lease expires (ORB-1101) can run again. Make side effects idempotent; see `troubleshooting-duplicates.md`.
[turn 1 · input 742 tok]
you> Is Python 3.13 supported?
atlas> Not yet. Orbit is tested up to Python 3.12; 3.13 support is an open feature request.
[turn 2 · input 815 tok]
you> Does Orbit support Kafka?
atlas> I don't know: the facts I have don't mention Kafka. Check the broker pages, starting with `concepts-brokers.md`.
[turn 3 · input 873 tok]
you> /history
  0 user      'My worker logs ORB-1101. Can a job run twice?'
  1 [reasoning item]
  2 assistant 'Yes. Orbit delivers jobs at least once, and a job whose lease' (phase=final_answer)
  ...
(illustrative; your numbers will differ)