Skip to content

symfonic.memory.backends.mongo_accepted_inbox

mongo_accepted_inbox

Private bounded discovery of ordered, authenticated native completions.

The counter and completion are committed in one Mongo transaction. A missing sequence is UNKNOWN, never a reason to run a model turn or skip a cadence tick.

accepted_batch async 

accepted_batch(journal: MongoOperationJournal, scope: MemoryScope, after: int, *, limit: int = 500) -> tuple[tuple[CompletedTurn, ...], bool]

Return at most limit consecutive completions and explicit backlog.

Source code in src/symfonic/memory/backends/mongo_accepted_inbox.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def accepted_batch(
    journal: MongoOperationJournal, scope: MemoryScope, after: int, *, limit: int = 500,
) -> tuple[tuple[CompletedTurn, ...], bool]:
    """Return at most limit consecutive completions and explicit backlog."""
    if (type(journal) is not MongoOperationJournal or type(scope) is not MemoryScope
            or scope.identity_version != 1 or type(after) is not int or after < 0
            or type(limit) is not int or not 1 <= limit <= 500):
        codec.refuse()
    try:
        await journal._topology()
        await journal._rows.create_index(
            [("scope", 1), ("completion.sequence", 1)],
            unique=True, name="accepted_scope_sequence",
            partialFilterExpression={"completion.sequence": {"$exists": True}})
        counter = journal._client[journal._database][
            "__symfonic_acceptance_sequence"].with_options(
                read_concern=journal._rows.read_concern,
                read_preference=journal._rows.read_preference)
        latest = await counter.find_one(
            {"_id": {"tenant": scope.tenant, "scope": scope.path}})
        highwater = 0 if latest is None else latest.get("value")
        if type(highwater) is not int or highwater < after:
            codec.refuse()
        rows = await journal._rows.find(
            {"_id.tenant": scope.tenant, "scope": scope.path,
             "completion.state": "ACCEPTED",
             "completion.sequence": {"$gt": after, "$lte": highwater}},
            {"_id": 1, "scope": 1, "request": 1, "completion.sequence": 1},
        ).sort("completion.sequence", 1).limit(limit + 1).to_list(length=limit + 1)
        batch = []
        for raw in rows[:limit]:
            identity, completion = raw.get("_id"), raw.get("completion")
            if (type(identity) is not dict or type(completion) is not dict
                    or raw.get("scope") != scope.path
                    or identity.get("tenant") != scope.tenant
                    or type(identity.get("invocation")) is not str
                    or completion.get("sequence") != after + len(batch) + 1):
                codec.refuse()
            accepted = await journal.read_completion(
                identity["invocation"], scope, raw.get("request"))
            if (type(accepted) is not CompletedTurn
                    or accepted.acceptance_sequence != completion["sequence"]):
                codec.refuse()
            batch.append(accepted)
        more = len(rows) > limit
        if (not more and after + len(batch) != highwater) or (
                more and after + len(batch) >= highwater):
            codec.refuse()
        return tuple(batch), more
    except MemoryContractError:
        raise
    except Exception as exc:
        journal._uncertain = True
        raise MemoryContractError("accepted inbox UNKNOWN") from exc