feat: implement Phase 1 capture and CI gate - #1
Conversation
Implement Event and AgentClock with thread-safe logical sequence allocation. Add Anthropic model-call capture, tool invocation/result capture, memory operations, and agent spawning with spawned_at_event_id. Add PostgreSQL storage, unit tests, integration tests, CI automation, local check scripts, and contributor documentation. Keep Phase 2+ functionality out of scope. Validation: 8 tests passed, Ruff passed, and ty passed.
DSCmatter
added
documentation
🤖 CodeAnt AI — Review Status
|
Thanks for using CodeAnt! 🎉We're free for open-source projects. if you're enjoying it, help us grow by sharing. Share on X · |
codeant-ai
Bot
added
the
size:XXL
![codeant-ai[bot]](https://avatars.githubusercontent.com/in/646884?s=60&v=4){kind=small}
| if existing_event is not None: | ||
| existing_agent = agent_store.get_by_spawn_event(existing_event.id) | ||
| if existing_agent is not None: | ||
| return existing_agent, existing_event, AgentClock(existing_agent.lamport_offset) |
There was a problem hiding this comment.
Suggestion: The retry path reconstructs a child clock from the persisted lamport_offset, but later child allocations are not written back to that record. After a retry or process restart, the returned clock can allocate sequence numbers already used by the child, causing duplicate logical sequences and PostgreSQL uniqueness failures. Persist or otherwise recover the child's latest logical sequence before returning a reconstructed clock. [stale reference]
Severity Level: Major ⚠️
- ❌ Retried child captures can fail PostgreSQL sequence uniqueness.
- ⚠️ Subsequent child tool or model events may not persist.
- ⚠️ In-memory retries also reconstruct stale counters.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** sdk/lifecycle.py
**Line:** 69:72
**Comment:**
*Stale Reference: The retry path reconstructs a child clock from the persisted `lamport_offset`, but later child allocations are not written back to that record. After a retry or process restart, the returned clock can allocate sequence numbers already used by the child, causing duplicate logical sequences and PostgreSQL uniqueness failures. Persist or otherwise recover the child's latest logical sequence before returning a reconstructed clock.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| def set(self, key: str, value: Any, *, causal_parent_ids: Iterable[str] = ()) -> None: | ||
| before = self.store.get(key, _MISSING) | ||
| self.store[key] = value | ||
| self._event( | ||
| "memory_write", | ||
| { | ||
| "operation": "set", | ||
| "key": key, | ||
| "before": None if before is _MISSING else before, |
There was a problem hiding this comment.
Suggestion: Memory mutation and event recording are separate unsynchronized operations. Concurrent writers can compute before from the same old value, overwrite each other, and emit events whose before/after metadata does not describe the actual serialized state transition. Protect the store mutation and corresponding event creation with one lock or use an atomic storage operation. [race condition]
Severity Level: Major ⚠️
- ⚠️ Shared memory writes can report incorrect transition history.
- ⚠️ Concurrent agents may overwrite each other's values.
- ⚠️ Replay or audit consumers receive inconsistent memory events.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** sdk/memory.py
**Line:** 58:66
**Comment:**
*Race Condition: Memory mutation and event recording are separate unsynchronized operations. Concurrent writers can compute `before` from the same old value, overwrite each other, and emit events whose before/after metadata does not describe the actual serialized state transition. Protect the store mutation and corresponding event creation with one lock or use an atomic storage operation.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| def _find_events(log: Any, invocation_id: str) -> list[Event]: | ||
| if hasattr(log, "events"): | ||
| events = log.events() | ||
| elif hasattr(log, "__iter__"): | ||
| events = list(log) | ||
| else: | ||
| events = [] | ||
| return [event for event in events if event.payload.get("invocation_id") == invocation_id] |
There was a problem hiding this comment.
Suggestion: Retry detection does not query stores that expose only get_by_idempotency_key, such as PostgresEventStore. _find_events therefore returns no prior events for PostgreSQL retries, so the wrapped tool executes again and repeats its side effects even though the subsequent event inserts are deduplicated. Query the store by agent and invocation idempotency key before invoking the tool. [api mismatch]
Severity Level: Major ⚠️
- ❌ PostgreSQL retries can repeat non-idempotent tool actions.
- ⚠️ Deduplicated events conceal the repeated execution.
- ⚠️ External side effects are not replay-safe.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** sdk/tools.py
**Line:** 15:22
**Comment:**
*Api Mismatch: Retry detection does not query stores that expose only `get_by_idempotency_key`, such as `PostgresEventStore`. `_find_events` therefore returns no prior events for PostgreSQL retries, so the wrapped tool executes again and repeats its side effects even though the subsequent event inserts are deduplicated. Query the store by agent and invocation idempotency key before invoking the tool.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| agent_id = kwargs.pop("agent_id", configured_agent_id) | ||
| clock = kwargs.pop("clock", configured_clock) | ||
| log = kwargs.pop("log", configured_log) | ||
| run_id = kwargs.pop("run_id", configured_run_id) | ||
| causal_parent_ids = kwargs.pop("causal_parent_ids", ()) | ||
| invocation_id = kwargs.pop("invocation_id", None) or str(uuid4()) |
There was a problem hiding this comment.
Suggestion: These reserved keyword removals make the decorator incompatible with tools that legitimately declare parameters named agent_id, clock, log, run_id, causal_parent_ids, or invocation_id: those arguments are consumed as capture context and are never passed to the wrapped function, causing missing arguments or altered tool behavior. Use a separate context mechanism rather than unconditionally removing names from the tool's call arguments. [api mismatch]
Severity Level: Major ⚠️
- ❌ Tools cannot receive reserved context-named inputs.
- ⚠️ Required tool parameters can raise `TypeError`.
- ⚠️ Valid tool calls may silently change behavior.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** sdk/tools.py
**Line:** 34:39
**Comment:**
*Api Mismatch: These reserved keyword removals make the decorator incompatible with tools that legitimately declare parameters named `agent_id`, `clock`, `log`, `run_id`, `causal_parent_ids`, or `invocation_id`: those arguments are consumed as capture context and are never passed to the wrapped function, causing missing arguments or altered tool behavior. Use a separate context mechanism rather than unconditionally removing names from the tool's call arguments.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| prior = _find_events(log, invocation_id) | ||
| prior_result = next((event for event in prior if event.event_type == "tool_result"), None) | ||
| if prior_result is not None: | ||
| return prior_result.payload.get("output") | ||
| prior_error = next((event for event in prior if event.event_type == "agent_error"), None) | ||
| if prior_error is not None: | ||
| raise RuntimeError(prior_error.payload.get("error", "captured tool failed")) |
There was a problem hiding this comment.
Suggestion: Prior events are matched only by invocation_id, so a shared log can return another agent's result or error when two agents reuse the same invocation identifier. Include the current agent_id and, where applicable, run_id in the lookup criteria to match the persistence idempotency scope. [incorrect condition logic]
Severity Level: Major ⚠️
- ❌ Shared logs can return another agent's tool output.
- ⚠️ Agents may skip required tool execution.
- ⚠️ Agent errors can incorrectly abort unrelated work.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** sdk/tools.py
**Line:** 43:49
**Comment:**
*Incorrect Condition Logic: Prior events are matched only by `invocation_id`, so a shared log can return another agent's result or error when two agents reuse the same invocation identifier. Include the current `agent_id` and, where applicable, `run_id` in the lookup criteria to match the persistence idempotency scope.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix
| latency_ms = int((time.monotonic() - start) * 1000) | ||
| record_event( | ||
| agent_id=self._owner.agent_id, | ||
| clock=self._owner.clock, | ||
| log=self._owner.log, | ||
| event_type="model_call", | ||
| payload={ | ||
| "model": kwargs.get("model"), | ||
| "input": kwargs.get("messages"), | ||
| "output": _dump_response(response), | ||
| "latency_ms": latency_ms, |
There was a problem hiding this comment.
Suggestion: When messages.create is called with Anthropic streaming enabled, the returned object is a stream that is consumed after this method returns. Calling _dump_response immediately records the stream object's representation instead of the generated message, and errors occurring during iteration are never captured as agent_error events. The wrapper must finalize or wrap the stream so capture occurs when consumption completes. [api mismatch]
Severity Level: Major ⚠️
- ❌ Streaming outputs are absent from model-call events.
- ⚠️ Streaming iteration failures are not captured.
- ⚠️ Recorded latency excludes stream consumption time.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** sdk/client.py
**Line:** 54:64
**Comment:**
*Api Mismatch: When `messages.create` is called with Anthropic streaming enabled, the returned object is a stream that is consumed after this method returns. Calling `_dump_response` immediately records the stream object's representation instead of the generated message, and errors occurring during iteration are never captured as `agent_error` events. The wrapper must finalize or wrap the stream so capture occurs when consumption completes.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| id uuid PRIMARY KEY DEFAULT gen_random_uuid(), | ||
| run_id uuid NOT NULL REFERENCES runs(id), | ||
| agent_id uuid NOT NULL REFERENCES agents(id), | ||
| logical_seq bigint NOT NULL, | ||
| wall_time timestamptz NOT NULL DEFAULT now(), | ||
| event_type event_type NOT NULL, | ||
| causal_parent_ids uuid[] NOT NULL DEFAULT '{}', |
There was a problem hiding this comment.
Suggestion: These PostgreSQL columns require UUID values, but the SDK accepts arbitrary strings for Event.id, run_id, agent_id, and causal_parent_ids. Events using valid in-memory identifiers such as planner or parent-1 will fail with a PostgreSQL UUID cast error. Align the schema with the SDK's string contract or validate and convert identifiers before persistence. [type error]
Severity Level: Major ⚠️
- ❌ PostgreSQL capture fails for non-UUID SDK identifiers.
- ❌ Events cannot persist with default or local-style IDs.
- ⚠️ In-memory and PostgreSQL backends have incompatible contracts.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** storage/postgres.py
**Line:** 43:49
**Comment:**
*Type Error: These PostgreSQL columns require UUID values, but the SDK accepts arbitrary strings for `Event.id`, `run_id`, `agent_id`, and `causal_parent_ids`. Events using valid in-memory identifiers such as `planner` or `parent-1` will fail with a PostgreSQL UUID cast error. Align the schema with the SDK's string contract or validate and convert identifiers before persistence.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| (event.agent_id, event.idempotency_key), | ||
| ) | ||
| row = cursor.fetchone() | ||
| self.connection.commit() |
There was a problem hiding this comment.
Suggestion: The insert can fail on the separate unique (agent_id, logical_seq) constraint, but this method has no rollback path before returning or propagating the database exception. A caller that catches the exception and reuses the connection will encounter PostgreSQL's aborted-transaction state on every subsequent operation. Roll back the connection on append failures before propagating the error. [possible bug]
Severity Level: Major ⚠️
- ❌ One failed append poisons subsequent connection operations.
- ⚠️ Retry and recovery workflows require external rollback.
- ⚠️ Shared worker connections may remain unusable.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** storage/postgres.py
**Line:** 115:115
**Comment:**
*Possible Bug: The insert can fail on the separate unique `(agent_id, logical_seq)` constraint, but this method has no rollback path before returning or propagating the database exception. A caller that catches the exception and reuses the connection will encounter PostgreSQL's aborted-transaction state on every subsequent operation. Roll back the connection on append failures before propagating the error.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fixRecover the latest child-agent logical sequence when handling spawn retries. Synchronize memory mutations with their captured before/after events and scope tool retry lookups by agent and idempotency key. Preserve wrapped-tool parameters when capture context is configured, while supporting explicit capture overrides. Capture Anthropic streaming responses after consumption and record streaming failures. Validate UUID identifiers before PostgreSQL persistence and roll back failed event or agent transactions. Add regression tests covering retries, concurrency, streaming, UUID validation, transaction rollback, and reserved tool parameters. Validation: - 15 tests passed - Ruff passed - ty passed - Neon PostgreSQL integration passed
|
@CodeAnt-AI can you review this again? I have pushed those fixes and validated them |
|
Question: can you review this again? i have pushed those fixes and validated them Answer: 1. Tool idempotency is not safe for concurrent retriesFile: The retry protection works for sequential retries, but two concurrent calls with the same prior = _find_events(log, agent_id, invocation_id)
...
invoke = record_event(...)
result = fn(*args, **kwargs)
This violates the stated guarantee that repeated invocations are prevented from running twice. A lock or claim/lease mechanism is needed around the invocation lifecycle. For PostgreSQL, this likely requires an atomic claim table or a transaction/advisory lock keyed by Please add a concurrent retry test, for example using a barrier inside the tool and asserting that the underlying function's call count remains 2. Spawn retries without an explicit child ID are not idempotentFile: child_id = child_agent_id or str(uuid4())
spawn_key = idempotency_key or f"agent-spawn:{child_id}"When neither Consider either:
Also add a test covering the retry behavior when only The sequential tests and validation results are useful, but they do not exercise the main remaining race: concurrent duplicate tool invocations. |
- serialize duplicate tool invocations per agent and invocation ID - add PostgreSQL advisory locking for concurrent retries - require retry-stable identity when spawning agents - add concurrent tool and spawn retry regression tests - update PostgreSQL setup documentation - validate with unit tests, Neon integration, Ruff, and ty
|
@CodeAnt-AI can you review this again? I have pushed those fixes and validated them |
|
Question: can you review this again? i have pushed those fixes and validated them Answer: 1. PostgreSQL event idempotency does not make logical sequence allocation safe
sequence = next_seq(clock, causal_parent_seqs)
return _append(log, Event(...))The PostgreSQL schema enforces: UNIQUE (agent_id, logical_seq)However, two processes using separate This is especially relevant because the PostgreSQL tool lock serializes one invocation identity, not all event writes for an agent. Consider one of these approaches:
Without this, the storage layer does not fully guarantee the “no duplicate logical sequences within an agent” invariant under multi-process usage. 2.
|
- allocate PostgreSQL logical sequences atomically per agent - preserve persisted agent records across spawn conflicts - distinguish missing memory values from stored None - capture abandoned and cleanup-failed streams - add regression and PostgreSQL concurrency tests - validate with tests, Neon, Ruff, and ty
|
@CodeAnt-AI now all good? |
|
Question: now all good? Answer:
One additional concern:
|
1 participant
User description
Implement Event and AgentClock with thread-safe logical sequence allocation.
Add Anthropic model-call capture, tool invocation/result capture, memory operations, and agent spawning with spawned_at_event_id.
Add PostgreSQL storage, unit tests, integration tests, CI automation, local check scripts, and contributor documentation.
Keep Phase 2+ functionality out of scope.
Validation: 8 tests passed, Ruff passed, and ty passed.
CodeAnt-AI Description
Add Phase 1 event capture for multi-agent workflows
What Changed
Impact
✅ Traceable model, tool, and memory activity✅ Safe retries without duplicate tool execution✅ Reliable parent-child event relationships💡 Usage Guide
Checking Your Pull Request
Every time you make a pull request, our system automatically looks through it. We check for security issues, mistakes in how you're setting up your infrastructure, and common code problems. We do this to make sure your changes are solid and won't cause any trouble later.
Talking to CodeAnt AI
Got a question or need a hand with something in your pull request? You can easily get in touch with CodeAnt AI right here. Just type the following in a comment on your pull request, and replace "Your question here" with whatever you want to ask:
This lets you have a chat with CodeAnt AI about your pull request, making it easier to understand and improve your code.
Example
Preserve Org Learnings with CodeAnt
You can record team preferences so CodeAnt AI applies them in future reviews. Reply directly to the specific CodeAnt AI suggestion (in the same thread) and replace "Your feedback here" with your input:
This helps CodeAnt AI learn and adapt to your team's coding style and standards.
Example
Retrigger review
Ask CodeAnt AI to review the PR again, by typing:
Check Your Repository Health
To analyze the health of your code repository, visit our dashboard at https://app.codeant.ai. This tool helps you identify potential issues and areas for improvement in your codebase, ensuring your repository maintains high standards of code health.