Skip to content

fix: align EventEmitter with server db.Event struct #63

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
beonde merged 3 commits into from
May 11, 2026
Merged

fix: align EventEmitter with server db.Event struct #63

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
59df865
fix: align EventEmitter with server db.Event struct
beonde May 11, 2026
acc5375
test: update event tests for db.Event struct alignment
beonde May 11, 2026
2d905c3
fix: copy data dict to avoid mutating caller's dict
beonde May 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 28 additions & 18 deletions capiscio_sdk/events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,18 +136,20 @@ def emit(
if not self.enabled:
return False

# Build a flat event matching the registry's db.Event struct.
# See capiscio-server internal/db/models.go for JSON field tags.
event = {
"id": str(uuid.uuid4()),
"type": event_type,
"agentId": self.agent_id,
"traceId": correlation_id or str(uuid.uuid4()),
"eventType": event_type,
"severity": "info",
"timestamp": datetime.now(timezone.utc).isoformat(),
"data": data or {},
"payload": dict(data) if data else {},
}

if task_id:
event["taskId"] = task_id
if correlation_id:
event["correlationId"] = correlation_id
event["payload"]["task_id"] = task_id

with self._batch_lock:
self._batch.append(event)
Expand Down Expand Up @@ -189,22 +191,30 @@ def flush(self) -> bool:
"X-Capiscio-Registry-Key": self.api_key,
}

# Send batch
response = self._client.post(
f"{self.server_url}/v1/events",
headers=headers,
json={"events": events_to_send},
)
# The registry decodes the POST body as a flat db.Event struct.
# Send each event individually.
failed: list = []
for event in events_to_send:
try:
response = self._client.post(
f"{self.server_url}/v1/events",
headers=headers,
json=event,
)
Comment on lines +194 to +203
Copy link
Copy Markdown
Member Author

@beonde beonde May 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Already fixed in acc5375 — tests updated to assert individual POSTs and new field names.

Comment on lines +194 to +203
Copy link
Copy Markdown
Member Author

@beonde beonde May 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Valid concern. In practice, the registry responds in <100ms so 10 sequential POSTs complete in ~1s. The 10s timeout is inherited from the httpx client default and only applies on failure. That said, adding a per-event timeout (e.g., 2s) and/or concurrent sends via a thread pool would be good hardening. Will track for a follow-up — not blocking for this fix since the previous batch approach had the same 10s single-request timeout and would also block on a slow/unreachable server.

if response.status_code not in (200, 201, 202):
logger.warning(f"Failed to send event: {response.status_code} {response.text[:200]}")
failed.append(event)
except Exception as e:
logger.error(f"Error sending event: {e}")
failed.append(event)

if response.status_code in (200, 201, 202):
logger.debug(f"Sent {len(events_to_send)} events")
return True
else:
logger.warning(f"Failed to send events: {response.status_code} {response.text}")
# Re-queue events on failure
if failed:
with self._batch_lock:
self._batch.extend(events_to_send)
self._batch.extend(failed)
return False

logger.debug(f"Sent {len(events_to_send)} events")
return True

except Exception as e:
logger.error(f"Error sending events: {e}")
Expand Down
33 changes: 17 additions & 16 deletions tests/unit/test_events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,10 +95,12 @@ def test_emit_adds_to_batch(self):
assert len(emitter._batch) == 1

event = emitter._batch[0]
assert event["type"] == "test_event"
assert event["eventType"] == "test_event"
assert event["agentId"] == "test-agent"
assert event["data"] == {"key": "value"}
assert event["payload"] == {"key": "value"}
assert event["severity"] == "info"
assert "id" in event
assert "traceId" in event
assert "timestamp" in event

def test_emit_with_task_id(self):
Expand All @@ -111,7 +113,7 @@ def test_emit_with_task_id(self):

emitter.emit("test_event", {}, task_id="task-123")

assert emitter._batch[0]["taskId"] == "task-123"
assert emitter._batch[0]["payload"]["task_id"] == "task-123"

def test_emit_with_correlation_id(self):
"""Test emit with correlation_id."""
Expand All @@ -123,7 +125,7 @@ def test_emit_with_correlation_id(self):

emitter.emit("test_event", {}, correlation_id="corr-456")

assert emitter._batch[0]["correlationId"] == "corr-456"
assert emitter._batch[0]["traceId"] == "corr-456"

def test_emit_flushes_when_batch_full(self):
"""Test that emit flushes when batch is full."""
Expand Down Expand Up @@ -175,8 +177,8 @@ def test_flush_sends_events(self):

# Add events directly to batch
emitter._batch = [
{"id": "1", "type": "event1", "data": {}},
{"id": "2", "type": "event2", "data": {}},
{"id": "1", "eventType": "event1", "payload": {}},
{"id": "2", "eventType": "event2", "payload": {}},
]

mock_response = MagicMock()
Expand All @@ -187,14 +189,13 @@ def test_flush_sends_events(self):

assert result is True
assert len(emitter._batch) == 0
mock_post.assert_called_once()
assert mock_post.call_count == 2 # Individual POST per event

call_kwargs = mock_post.call_args[1]
assert call_kwargs["json"]["events"] == [
{"id": "1", "type": "event1", "data": {}},
{"id": "2", "type": "event2", "data": {}},
]
assert call_kwargs["headers"]["X-Capiscio-Registry-Key"] == "sk_test"
# Verify each event sent individually
calls = mock_post.call_args_list
assert calls[0][1]["json"] == {"id": "1", "eventType": "event1", "payload": {}}
assert calls[1][1]["json"] == {"id": "2", "eventType": "event2", "payload": {}}
assert calls[0][1]["headers"]["X-Capiscio-Registry-Key"] == "sk_test"

def test_flush_requeues_on_failure(self):
"""Test that flush requeues events on server error."""
Expand All @@ -204,7 +205,7 @@ def test_flush_requeues_on_failure(self):
agent_id="test-agent",
)

events = [{"id": "1", "type": "event1", "data": {}}]
events = [{"id": "1", "eventType": "event1", "payload": {}}]
emitter._batch = events.copy()

mock_response = MagicMock()
Expand All @@ -225,7 +226,7 @@ def test_flush_requeues_on_exception(self):
agent_id="test-agent",
)

emitter._batch = [{"id": "1", "type": "event1", "data": {}}]
emitter._batch = [{"id": "1", "eventType": "event1", "payload": {}}]

with patch.object(emitter._client, "post", side_effect=Exception("Network error")):
result = emitter.flush()
Expand All @@ -243,7 +244,7 @@ def test_flush_disabled(self):
)

# Manually add event (bypassing emit's disabled check)
emitter._batch = [{"id": "1", "type": "event1", "data": {}}]
emitter._batch = [{"id": "1", "eventType": "event1", "payload": {}}]

result = emitter.flush()

Expand Down
Loading