Route a frame by id, branch on kind (report 13 §4.1).
(self, env: pb.Envelope)
| 158 | self._fail_all(CmdopConnectionError(f"core read loop failed: {exc}")) |
| 159 | |
| 160 | def _dispatch(self, env: pb.Envelope) -> None: |
| 161 | """Route a frame by id, branch on kind (report 13 §4.1).""" |
| 162 | p = self._pending.get(env.id) |
| 163 | if p is None: |
| 164 | return |
| 165 | kind = env.kind |
| 166 | |
| 167 | if isinstance(p, _UnaryPending): |
| 168 | if kind == pb.Envelope.KIND_ERROR: |
| 169 | self._pending.pop(env.id, None) |
| 170 | if not p.fut.done(): |
| 171 | p.fut.set_exception(self._to_error(env)) |
| 172 | elif kind == pb.Envelope.KIND_RESPONSE: |
| 173 | self._pending.pop(env.id, None) |
| 174 | if not p.fut.done(): |
| 175 | p.fut.set_result(env) |
| 176 | # Any other kind on a unary id is a protocol violation — ignore. |
| 177 | return |
| 178 | |
| 179 | # Streaming call (machines.ask): EVENT/CALLBACK push; DONE/ERROR/RESPONSE |
| 180 | # terminate. |
| 181 | queue = p.queue |
| 182 | if kind in (pb.Envelope.KIND_EVENT, pb.Envelope.KIND_CALLBACK): |
| 183 | queue.put_nowait(env) |
| 184 | elif kind == pb.Envelope.KIND_DONE: |
| 185 | self._pending.pop(env.id, None) |
| 186 | queue.put_nowait(env) |
| 187 | queue.put_nowait(_STREAM_END) |
| 188 | elif kind == pb.Envelope.KIND_ERROR: |
| 189 | # Push the raw ERROR envelope; FrameStream raises it as an |
| 190 | # AgentStreamError (the ask stream's error-frame semantics, mirroring |
| 191 | # the archived wrapper) rather than the unary code->exception map. |
| 192 | self._pending.pop(env.id, None) |
| 193 | queue.put_nowait(env) |
| 194 | queue.put_nowait(_STREAM_END) |
| 195 | elif kind == pb.Envelope.KIND_RESPONSE: |
| 196 | # A unary-shaped reply on a stream id (shouldn't happen for ask, but |
| 197 | # be forgiving): deliver then end. |
| 198 | self._pending.pop(env.id, None) |
| 199 | queue.put_nowait(env) |
| 200 | queue.put_nowait(_STREAM_END) |
| 201 | |
| 202 | @staticmethod |
| 203 | def _to_error(env: pb.Envelope) -> Exception: |
no test coverage detected