<!-- Sema documentation — autogen-arithmetic-agent
     Sema is a new AI-native, neurosymbolic programming language most models
     have not been trained on. Full corpus: https://sema.49.12.246.95.sslip.io/llms-full.txt
     Install language support: https://sema.49.12.246.95.sslip.io/install-skill/ -->

# autogen-arithmetic-agent

> The autogen-arithmetic-agent worked example.

> The autogen-arithmetic-agent worked example.

Run it from `sema/`:

```bash
sema check examples/autogen-arithmetic-agent
SEMA_STRICT=1 sema run examples/autogen-arithmetic-agent
sema assure examples/autogen-arithmetic-agent --grade silver
```

## Source

### `src/main.sema`

```sema
"""Bounded AutoGen arithmetic-team parity with persisted typed state."""

assure silver

enum OutcomeKind:
    success | failure

struct Request:
    lhs: int
    rhs: int

struct Outcome:
    kind: OutcomeKind
    value: Option[int]
    error: Option[str]
    invariant (kind == OutcomeKind.success and value != None and error == None) or (kind == OutcomeKind.failure and value == None and error != None)

struct Candidate:
    outcome: Outcome
    content: str

struct Verdict:
    outcome: Outcome
    content: str

struct TeamState:
    request: Request
    candidate: Candidate
    messages: list[dict[str, any]]
    next_speaker: str
    generation: int
    invariant len(messages) == 2
    invariant next_speaker == "verifier"
    invariant generation >= 0

struct ResetTeam:
    request: Request
    messages: list[dict[str, any]]
    next_speaker: str
    generation: int
    invariant len(messages) == 0
    invariant next_speaker == "solver"
    invariant generation >= 0

struct UsedTeam:
    request: Request
    messages: list[dict[str, any]]
    generation: int
    invariant len(messages) == 3
    invariant generation >= 0

struct Execution:
    view: dict[str, any]
    used: UsedTeam

struct ResumeExecution:
    execution: Execution
    midpoint_message_count: int
    duplicate_count: int
    invariant midpoint_message_count == 2
    invariant duplicate_count >= 0

def success(value: int) -> Outcome !{}:
    return Outcome(kind=OutcomeKind.success, value=Some(value), error=None)

def failure(name: str) -> Outcome !{}:
    require len(name) > 0
    return Outcome(kind=OutcomeKind.failure, value=None, error=Some(name))

def outcome_value(outcome: Outcome) -> int !{}:
    match outcome.value:
        case Some(value):
            return value
        case None:
            return 0

def outcome_error(outcome: Outcome) -> str !{}:
    match outcome.error:
        case Some(error):
            return error
        case None:
            return ""

def outcome_tag(outcome: Outcome) -> str !{}:
    return "ok" if outcome.kind == OutcomeKind.success else "error"

def candidate_for(request: Request) -> Candidate !{}:
    if request.rhs == 0:
        outcome = failure("ZeroDivisionError")
        return Candidate(outcome=outcome, content="candidate:error:" + outcome_error(outcome))
    outcome = success(request.lhs // request.rhs)
    return Candidate(outcome=outcome, content="candidate:ok:" + str(outcome_value(outcome)))

def verdict_for(candidate: Candidate) -> Verdict !{}:
    if candidate.outcome.kind == OutcomeKind.failure:
        return Verdict(outcome=candidate.outcome,
                       content="final:error:" + outcome_error(candidate.outcome) + ":TERMINATE")
    return Verdict(outcome=candidate.outcome,
                   content="final:ok:" + str(outcome_value(candidate.outcome)) + ":TERMINATE")

@provides("agent.execute")
def scripted_model(packet: dict[str, any]) -> any !{}:
    if packet["agent"] == "solver":
        return candidate_for(packet["inputs"]["request"])
    if packet["agent"] == "verifier":
        return verdict_for(packet["inputs"]["candidate"])
    return None

agent solver(request: Request) -> Candidate by arithmetic_model:
    sem "Compute exact signed floor division or return a typed zero-division outcome"
    budget model_calls=1, tokens=64

agent verifier(candidate: Candidate) -> Verdict by verifier_model:
    sem "Verify the candidate and emit the bounded terminal verdict"
    budget model_calls=1, tokens=64

def user_event(request: Request) -> dict[str, any] !{}:
    return {"ordinal": 0, "source": "user", "kind": "TextMessage",
            "content": "floor_div " + str(request.lhs) + " " + str(request.rhs)}

def agent_event(ordinal: int, source: str, content: str) -> dict[str, any] !{}:
    return {"ordinal": ordinal, "source": source, "kind": "TextMessage", "content": content}

def outcome_payload(outcome: Outcome) -> dict[str, any] !{}:
    if outcome.kind == OutcomeKind.failure:
        return {"kind": "error", "value": None, "error": outcome_error(outcome)}
    return {"kind": "ok", "value": outcome_value(outcome), "error": None}

def outcome_from_payload(raw: dict[str, any]) -> Outcome !{}:
    require raw["kind"] == "ok" or raw["kind"] == "error"
    if raw["kind"] == "ok":
        ensure raw["value"] != None
        ensure raw["error"] == None
        return success(raw["value"])
    ensure raw["value"] == None
    ensure raw["error"] != None
    ensure len(raw["error"]) > 0
    return failure(raw["error"])

def save_midpoint(request: Request, candidate: Candidate, generation: int) -> TeamState !{}:
    return TeamState(request=request, candidate=candidate,
                     messages=[user_event(request), agent_event(1, "solver", candidate.content)],
                     next_speaker="verifier", generation=generation)

def serialize_midpoint(state: TeamState) -> str !{}:
    return json.dumps({"request": {"lhs": state.request.lhs, "rhs": state.request.rhs},
                       "candidate": {"outcome": outcome_payload(state.candidate.outcome),
                                     "content": state.candidate.content},
                       "messages": state.messages, "next_speaker": state.next_speaker,
                       "generation": state.generation})

def loaded_event(raw: dict[str, any], ordinal: int, source: str) -> dict[str, any] !{}:
    require raw["ordinal"] == ordinal
    require raw["source"] == source
    require raw["kind"] == "TextMessage"
    require len(raw["content"]) > 0
    return agent_event(ordinal, source, raw["content"])

def load_midpoint(encoded: str) -> TeamState !{}:
    raw = json.loads(encoded)
    ensure raw["next_speaker"] == "verifier"
    ensure raw["generation"] >= 0
    request = Request(lhs=raw["request"]["lhs"], rhs=raw["request"]["rhs"])
    outcome = outcome_from_payload(raw["candidate"]["outcome"])
    candidate = Candidate(outcome=outcome, content=raw["candidate"]["content"])
    ensure candidate.content == candidate_for(request).content
    ensure len(raw["messages"]) == 2
    first = loaded_event(raw["messages"][0], 0, "user")
    second = loaded_event(raw["messages"][1], 1, "solver")
    ensure first == user_event(request)
    ensure second == agent_event(1, "solver", candidate.content)
    return TeamState(request=request, candidate=candidate, messages=[first, second],
                     next_speaker="verifier", generation=raw["generation"])

def final_payload(outcome: Outcome) -> dict[str, any] !{}:
    if outcome.kind == OutcomeKind.failure:
        return {"status": "error", "value": None, "error": outcome_error(outcome)}
    return {"status": "ok", "value": outcome_value(outcome), "error": None}

def agent_message_count(messages: list[dict[str, any]]) -> int !{}:
    mut count = 0
    for message in messages:
        if message["source"] == "solver" or message["source"] == "verifier":
            count = count + 1
    return count

def duplicate_messages(prior: list[dict[str, any]], resumed: list[dict[str, any]]) -> int !{}:
    mut count = 0
    for later in resumed:
        for earlier in prior:
            if later["source"] == earlier["source"] and later["content"] == earlier["content"]:
                count = count + 1
    return count

def run_view(messages: list[dict[str, any]], verdict: Verdict) -> dict[str, any] !{}:
    return {"events": messages, "final": final_payload(verdict.outcome),
            "stop_reason": "text_mention", "message_count": len(messages),
            "model_calls": agent_message_count(messages)}

def initial_team(request: Request) -> ResetTeam !{}:
    return ResetTeam(request=request, messages=[], next_speaker="solver", generation=0)

def execute_from_start(team: ResetTeam) -> Execution !{model.invoke}:
    require team.next_speaker == "solver"
    candidate = solver(team.request)
    verdict = verifier(candidate)
    messages = [user_event(team.request), agent_event(1, "solver", candidate.content),
                agent_event(2, "verifier", verdict.content)]
    return Execution(view=run_view(messages, verdict),
                     used=UsedTeam(request=team.request, messages=messages,
                                   generation=team.generation))

def uninterrupted(request: Request) -> Execution !{model.invoke}:
    return execute_from_start(initial_team(request))

def resumed(request: Request) -> ResumeExecution !{model.invoke}:
    candidate = solver(request)
    encoded = serialize_midpoint(save_midpoint(request, candidate, 0))
    loaded = load_midpoint(encoded)
    verdict = verifier(loaded.candidate)
    resumed_messages = [agent_event(2, loaded.next_speaker, verdict.content)]
    midpoint_count = len(loaded.messages)
    duplicates = duplicate_messages(loaded.messages, resumed_messages)
    mut messages = loaded.messages
    messages.append(resumed_messages[0])
    execution = Execution(view=run_view(messages, verdict),
                          used=UsedTeam(request=loaded.request, messages=messages,
                                        generation=loaded.generation))
    return ResumeExecution(execution=execution, midpoint_message_count=midpoint_count,
                           duplicate_count=duplicates)

def reset_team(used: UsedTeam) -> ResetTeam !{}:
    return ResetTeam(request=Request(lhs=used.request.lhs, rhs=used.request.rhs),
                     messages=[], next_speaker="solver", generation=used.generation + 1)

def rerun_after_reset(used: UsedTeam) -> Execution !{model.invoke}:
    return execute_from_start(reset_team(used))

circuit case_paths(request: Request) -> dict[str, any] !{model.invoke}:
    budget agents=6, spawn_depth=0, model_calls=6, tokens=512
    full = uninterrupted(request)
    restored = resumed(request)
    reset = rerun_after_reset(full.used)
    ensure full.view == restored.execution.view
    ensure full.view == reset.view
    ensure reset.used.generation == full.used.generation + 1
    save_load_equal = full.view == restored.execution.view
    reset_equal = full.view == reset.view
    return {"id": str(request.lhs) + "//" + str(request.rhs),
            "input": {"lhs": request.lhs, "rhs": request.rhs},
            "uninterrupted": full.view, "resumed": restored.execution.view, "reset": reset.view,
            "midpoint_stop_reason": "max_messages",
            "midpoint_message_count": restored.midpoint_message_count,
            "resume_duplicate_count": restored.duplicate_count,
            "save_load_equal": save_load_equal, "reset_equal": reset_equal}

def corpus() -> dict[str, any] !{model.invoke}:
    inputs = [Request(lhs=7, rhs=3), Request(lhs=-7, rhs=3),
              Request(lhs=7, rhs=-3), Request(lhs=-7, rhs=-3),
              Request(lhs=0, rhs=5), Request(lhs=5, rhs=0)]
    mut cases = []
    for request in inputs:
        cases.append(case_paths(request))
    return {"schema": "autogen-arithmetic-parity/v1", "cases": cases}

test "signed floor division uses Python semantics":
    ensure outcome_value(candidate_for(Request(lhs=7, rhs=3)).outcome) == 2
    ensure outcome_value(candidate_for(Request(lhs=-7, rhs=3)).outcome) == -3
    ensure outcome_value(candidate_for(Request(lhs=7, rhs=-3)).outcome) == -3
    ensure outcome_value(candidate_for(Request(lhs=-7, rhs=-3)).outcome) == 2

test "division by zero is a mutually exclusive typed outcome":
    outcome = candidate_for(Request(lhs=5, rhs=0)).outcome
    ensure outcome.kind == OutcomeKind.failure
    ensure outcome.value == None
    ensure outcome_error(outcome) == "ZeroDivisionError"

test "midpoint crosses JSON and reset consumes used state":
    request = Request(lhs=-7, rhs=3)
    saved = save_midpoint(request, candidate_for(request), 0)
    encoded = serialize_midpoint(saved)
    loaded = load_midpoint(encoded)
    ensure serialize_midpoint(loaded) == encoded
    used = UsedTeam(request=request,
                    messages=[user_event(request), agent_event(1, "solver", saved.candidate.content),
                              agent_event(2, "verifier", verdict_for(saved.candidate).content)],
                    generation=0)
    reset = reset_team(used)
    ensure reset.generation == 1
    ensure len(reset.messages) == 0
    ensure reset.next_speaker == "solver"

def main() -> str !{model.invoke}:
    return json.dumps(corpus())
```

## Reflected API

# `main`

Bounded AutoGen arithmetic-team parity with persisted typed state.

# `enum OutcomeKind`

**Variants**

- `success`
- `failure`

# `struct Request`

**Fields**

| field | type | descriptor |
|---|---|---|
| `lhs` | `int` |  |
| `rhs` | `int` |  |

# `struct Outcome`

**Fields**

| field | type | descriptor |
|---|---|---|
| `kind` | `OutcomeKind` |  |
| `value` | `Option[int]` |  |
| `error` | `Option[str]` |  |

# `struct Candidate`

**Fields**

| field | type | descriptor |
|---|---|---|
| `outcome` | `Outcome` |  |
| `content` | `str` |  |

# `struct Verdict`

**Fields**

| field | type | descriptor |
|---|---|---|
| `outcome` | `Outcome` |  |
| `content` | `str` |  |

# `struct TeamState`

**Fields**

| field | type | descriptor |
|---|---|---|
| `request` | `Request` |  |
| `candidate` | `Candidate` |  |
| `messages` | `list[dict[str, any]]` |  |
| `next_speaker` | `str` |  |
| `generation` | `int` |  |

# `struct ResetTeam`

**Fields**

| field | type | descriptor |
|---|---|---|
| `request` | `Request` |  |
| `messages` | `list[dict[str, any]]` |  |
| `next_speaker` | `str` |  |
| `generation` | `int` |  |

# `struct UsedTeam`

**Fields**

| field | type | descriptor |
|---|---|---|
| `request` | `Request` |  |
| `messages` | `list[dict[str, any]]` |  |
| `generation` | `int` |  |

# `struct Execution`

**Fields**

| field | type | descriptor |
|---|---|---|
| `view` | `dict[str, any]` |  |
| `used` | `UsedTeam` |  |

# `struct ResumeExecution`

**Fields**

| field | type | descriptor |
|---|---|---|
| `execution` | `Execution` |  |
| `midpoint_message_count` | `int` |  |
| `duplicate_count` | `int` |  |

# `def success`

```sema
def success(value: int) -> Outcome !{}
```

**Parameters**

| name | type |
|---|---|
| `value` | `int` |

**Returns** `Outcome`

**Effects** `!{}`

# `def failure`

```sema
def failure(name: str) -> Outcome !{}
```

**Parameters**

| name | type |
|---|---|
| `name` | `str` |

**Returns** `Outcome`

**Effects** `!{}`

# `def outcome_value`

```sema
def outcome_value(outcome: Outcome) -> int !{}
```

**Parameters**

| name | type |
|---|---|
| `outcome` | `Outcome` |

**Returns** `int`

**Effects** `!{}`

# `def outcome_error`

```sema
def outcome_error(outcome: Outcome) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `outcome` | `Outcome` |

**Returns** `str`

**Effects** `!{}`

# `def outcome_tag`

```sema
def outcome_tag(outcome: Outcome) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `outcome` | `Outcome` |

**Returns** `str`

**Effects** `!{}`

# `def candidate_for`

```sema
def candidate_for(request: Request) -> Candidate !{}
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |

**Returns** `Candidate`

**Effects** `!{}`

# `def verdict_for`

```sema
def verdict_for(candidate: Candidate) -> Verdict !{}
```

**Parameters**

| name | type |
|---|---|
| `candidate` | `Candidate` |

**Returns** `Verdict`

**Effects** `!{}`

# `def scripted_model`

```sema
def scripted_model(packet: dict[str, any]) -> any !{}
```

**Parameters**

| name | type |
|---|---|
| `packet` | `dict[str, any]` |

**Returns** `any`

**Effects** `!{}`

# `agent solver`

```sema
agent solver(request: Request) -> Candidate
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |

**Returns** `Candidate`

# `agent verifier`

```sema
agent verifier(candidate: Candidate) -> Verdict
```

**Parameters**

| name | type |
|---|---|
| `candidate` | `Candidate` |

**Returns** `Verdict`

# `def user_event`

```sema
def user_event(request: Request) -> dict[str, any] !{}
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |

**Returns** `dict[str, any]`

**Effects** `!{}`

# `def agent_event`

```sema
def agent_event(ordinal: int, source: str, content: str) -> dict[str, any] !{}
```

**Parameters**

| name | type |
|---|---|
| `ordinal` | `int` |
| `source` | `str` |
| `content` | `str` |

**Returns** `dict[str, any]`

**Effects** `!{}`

# `def outcome_payload`

```sema
def outcome_payload(outcome: Outcome) -> dict[str, any] !{}
```

**Parameters**

| name | type |
|---|---|
| `outcome` | `Outcome` |

**Returns** `dict[str, any]`

**Effects** `!{}`

# `def outcome_from_payload`

```sema
def outcome_from_payload(raw: dict[str, any]) -> Outcome !{}
```

**Parameters**

| name | type |
|---|---|
| `raw` | `dict[str, any]` |

**Returns** `Outcome`

**Effects** `!{}`

# `def save_midpoint`

```sema
def save_midpoint(request: Request, candidate: Candidate, generation: int) -> TeamState !{}
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |
| `candidate` | `Candidate` |
| `generation` | `int` |

**Returns** `TeamState`

**Effects** `!{}`

# `def serialize_midpoint`

```sema
def serialize_midpoint(state: TeamState) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `state` | `TeamState` |

**Returns** `str`

**Effects** `!{}`

# `def loaded_event`

```sema
def loaded_event(raw: dict[str, any], ordinal: int, source: str) -> dict[str, any] !{}
```

**Parameters**

| name | type |
|---|---|
| `raw` | `dict[str, any]` |
| `ordinal` | `int` |
| `source` | `str` |

**Returns** `dict[str, any]`

**Effects** `!{}`

# `def load_midpoint`

```sema
def load_midpoint(encoded: str) -> TeamState !{}
```

**Parameters**

| name | type |
|---|---|
| `encoded` | `str` |

**Returns** `TeamState`

**Effects** `!{}`

# `def final_payload`

```sema
def final_payload(outcome: Outcome) -> dict[str, any] !{}
```

**Parameters**

| name | type |
|---|---|
| `outcome` | `Outcome` |

**Returns** `dict[str, any]`

**Effects** `!{}`

# `def agent_message_count`

```sema
def agent_message_count(messages: list[dict[str, any]]) -> int !{}
```

**Parameters**

| name | type |
|---|---|
| `messages` | `list[dict[str, any]]` |

**Returns** `int`

**Effects** `!{}`

# `def duplicate_messages`

```sema
def duplicate_messages(prior: list[dict[str, any]], resumed: list[dict[str, any]]) -> int !{}
```

**Parameters**

| name | type |
|---|---|
| `prior` | `list[dict[str, any]]` |
| `resumed` | `list[dict[str, any]]` |

**Returns** `int`

**Effects** `!{}`

# `def run_view`

```sema
def run_view(messages: list[dict[str, any]], verdict: Verdict) -> dict[str, any] !{}
```

**Parameters**

| name | type |
|---|---|
| `messages` | `list[dict[str, any]]` |
| `verdict` | `Verdict` |

**Returns** `dict[str, any]`

**Effects** `!{}`

# `def initial_team`

```sema
def initial_team(request: Request) -> ResetTeam !{}
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |

**Returns** `ResetTeam`

**Effects** `!{}`

# `def execute_from_start`

```sema
def execute_from_start(team: ResetTeam) -> Execution !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `team` | `ResetTeam` |

**Returns** `Execution`

**Effects** `!{model.invoke}`

# `def uninterrupted`

```sema
def uninterrupted(request: Request) -> Execution !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |

**Returns** `Execution`

**Effects** `!{model.invoke}`

# `def resumed`

```sema
def resumed(request: Request) -> ResumeExecution !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |

**Returns** `ResumeExecution`

**Effects** `!{model.invoke}`

# `def reset_team`

```sema
def reset_team(used: UsedTeam) -> ResetTeam !{}
```

**Parameters**

| name | type |
|---|---|
| `used` | `UsedTeam` |

**Returns** `ResetTeam`

**Effects** `!{}`

# `def rerun_after_reset`

```sema
def rerun_after_reset(used: UsedTeam) -> Execution !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `used` | `UsedTeam` |

**Returns** `Execution`

**Effects** `!{model.invoke}`

# `circuit case_paths`

```sema
circuit case_paths(request: Request) -> dict[str, any] !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `request` | `Request` |

**Returns** `dict[str, any]`

**Effects** `!{model.invoke}`

# `def corpus`

```sema
def corpus() -> dict[str, any] !{model.invoke}
```

**Returns** `dict[str, any]`

**Effects** `!{model.invoke}`

# `def main`

```sema
def main() -> str !{model.invoke}
```

**Returns** `str`

**Effects** `!{model.invoke}`