Skip to content

ccproxy.plugins.codex.responses_state

ccproxy.plugins.codex.responses_state

Local OpenAI Responses continuation state for the Codex backend.

ResponsesStateNotFoundError 

ResponsesStateNotFoundError(response_id)

Bases: ValueError

Raised when a local previous_response_id cannot be resolved.

Source code in ccproxy/plugins/codex/responses_state.py 
23
24
25
def __init__(self, response_id: str) -> None:
    self.response_id = response_id
    super().__init__(f"Unknown previous_response_id: {response_id}")

CodexResponsesStateStore 

CodexResponsesStateStore(
    *, max_entries=1024, ttl_seconds=3600
)

Bounded per-client state used to emulate previous_response_id locally.

Source code in ccproxy/plugins/codex/responses_state.py 
49
50
51
52
53
54
55
def __init__(self, *, max_entries: int = 1024, ttl_seconds: int = 3600) -> None:
    self.max_entries = _positive_int(max_entries, default=1024)
    self.ttl_seconds = _positive_int(ttl_seconds, default=3600)
    self._records: OrderedDict[tuple[str, str], ResponsesStateRecord] = (
        OrderedDict()
    )
    self._lock = RLock()

prepare_payload 

prepare_payload(payload, *, headers)

Return a provider payload with any local continuation expanded.

Source code in ccproxy/plugins/codex/responses_state.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def prepare_payload(
    self,
    payload: dict[str, Any],
    *,
    headers: dict[str, str],
) -> tuple[dict[str, Any], str, str | None]:
    """Return a provider payload with any local continuation expanded."""

    scope = self.scope_for_headers(headers)
    previous_response_id = payload.get("previous_response_id")
    prepared = copy.deepcopy(payload)

    if previous_response_id is None:
        return prepared, scope, None

    if not isinstance(previous_response_id, str) or not previous_response_id:
        return prepared, scope, None

    record = self.get(scope, previous_response_id)
    if record is None:
        raise ResponsesStateNotFoundError(previous_response_id)

    current_input = _normalize_input_items(prepared.get("input", []))
    prepared["input"] = copy.deepcopy(record.context_items) + current_input
    prepared.pop("previous_response_id", None)
    return prepared, scope, previous_response_id

store_response 

store_response(*, scope, request_payload, response_payload)

Store a completed or tool-pending Responses payload for continuation.

Source code in ccproxy/plugins/codex/responses_state.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def store_response(
    self,
    *,
    scope: str,
    request_payload: dict[str, Any],
    response_payload: dict[str, Any],
) -> bool:
    """Store a completed or tool-pending Responses payload for continuation."""

    if not isinstance(response_payload, dict):
        return False

    response_id = response_payload.get("id")
    if not isinstance(response_id, str) or not response_id:
        return False

    status = response_payload.get("status")
    if isinstance(status, str) and status.lower() in _FAILED_STATUSES:
        return False

    output_items = response_payload.get("output")
    if not isinstance(output_items, list):
        output_items = []

    input_items = _normalize_input_items(request_payload.get("input", []))
    context_items = input_items + _normalize_output_items(output_items)
    if not context_items:
        return False

    expires_at = time.monotonic() + self.ttl_seconds
    record = ResponsesStateRecord(
        scope=scope,
        response_id=response_id,
        context_items=context_items,
        expires_at=expires_at,
    )
    key = (scope, response_id)
    with self._lock:
        self._prune_locked(now=time.monotonic())
        self._records[key] = record
        self._records.move_to_end(key)
        while len(self._records) > self.max_entries:
            self._records.popitem(last=False)
    return True