fix(tempo): only log red team execution time, use pre-computed duration
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled

Two bugs fixed:

1. Blue team evaluation was also sent to Tempo. Only operator (red team)
   execution time should be logged — blue team time is tracked internally
   in Aegis but does NOT represent billable operator work. Added a
   whitelist (_TEMPO_ACTIVITY_TYPES = {"red_team_execution"}).

2. _calculate_duration() re-computed duration from red_started_at to
   datetime.utcnow() at call time, without subtracting paused seconds.
   This caused inflated times (e.g. 45 min instead of 5 min) when there
   was any delay between the workflow transition and the Tempo call.
   Now the duration_seconds already computed by _create_phase_worklog
   (gross elapsed - paused) is passed directly to auto_log_test_worklog
   and used as-is, so Aegis and Tempo always agree on the duration.

Also: use red_started_at as the worklog date (not submission timestamp)
so the Tempo entry reflects when the work actually happened.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
kitos
2026-05-27 11:38:44 +02:00
parent 44ef4129a5
commit 0e6cec4d07
2 changed files with 49 additions and 39 deletions

View File

@@ -8,6 +8,13 @@ Obtain a Tempo token at: Jira → Apps → Tempo → Settings → API Integratio
The global ``settings.TEMPO_ENABLED`` flag acts as a kill-switch. When False, The global ``settings.TEMPO_ENABLED`` flag acts as a kill-switch. When False,
all Tempo calls are silently skipped regardless of whether users have tokens. all Tempo calls are silently skipped regardless of whether users have tokens.
What goes to Tempo
------------------
Only **red team execution** time is logged to Tempo. This reflects the time
the operator (red_tech) spends executing the attack technique — from the moment
they click "Start" to when they click "Done". Blue team evaluation time is
tracked internally in Aegis but is NOT sent to Tempo.
""" """
import logging import logging
@@ -21,6 +28,11 @@ from app.models.jira_link import JiraLink, JiraLinkEntityType
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Only these activity types are forwarded to Tempo.
# "blue_team_evaluation" is intentionally excluded — it is tracked in Aegis
# but does not represent operator execution work.
_TEMPO_ACTIVITY_TYPES = {"red_team_execution"}
def has_tempo_configured(user) -> bool: def has_tempo_configured(user) -> bool:
"""Return True if *user* has a personal Tempo API token stored.""" """Return True if *user* has a personal Tempo API token stored."""
@@ -78,16 +90,37 @@ def auto_log_test_worklog(
test, test,
user, user,
activity_type: str, activity_type: str,
duration_seconds: int,
) -> Optional[dict]: ) -> Optional[dict]:
"""If the test has a Jira link and *user* has a Tempo token, log time. """Log *duration_seconds* to Tempo for the given test if conditions are met.
Returns the Tempo worklog response, or None if skipped. ``duration_seconds`` must be the value already computed by the workflow
layer (gross elapsed time minus any paused time). It is used as-is so
the Tempo entry always matches the Aegis worklog — no re-calculation.
Only ``red_team_execution`` activities are forwarded to Tempo.
``blue_team_evaluation`` is tracked internally but not sent.
Returns the Tempo worklog response dict, or ``None`` if skipped.
Completely non-fatal — errors are logged and swallowed. Completely non-fatal — errors are logged and swallowed.
""" """
# Only operator execution time goes to Tempo
if activity_type not in _TEMPO_ACTIVITY_TYPES:
logger.debug(
"Skipping Tempo sync for activity_type=%s (not in whitelist)", activity_type
)
return None
# Global kill-switch # Global kill-switch
if not settings.TEMPO_ENABLED: if not settings.TEMPO_ENABLED:
return None return None
if duration_seconds <= 0:
logger.debug(
"Skipping Tempo sync for test %s: duration=%ds", test.id, duration_seconds
)
return None
# Per-user token required # Per-user token required
if not has_tempo_configured(user): if not has_tempo_configured(user):
logger.debug( logger.debug(
@@ -118,22 +151,24 @@ def auto_log_test_worklog(
) )
return None return None
duration = _calculate_duration(test, activity_type)
if duration <= 0:
return None
try: try:
# Use red_started_at date as the worklog date so it matches when the
# work actually happened (not the submission timestamp).
work_date = (
(test.red_started_at or getattr(test, "updated_at", None) or test.created_at)
.strftime("%Y-%m-%d")
)
result = log_worklog( result = log_worklog(
user=user, user=user,
jira_issue_id=int(link.jira_issue_id), jira_issue_id=int(link.jira_issue_id),
author_account_id=jira_account_id, author_account_id=jira_account_id,
date=(getattr(test, "updated_at", None) or test.created_at).strftime("%Y-%m-%d"), date=work_date,
time_spent_seconds=duration, time_spent_seconds=duration_seconds,
description=f"[Aegis] {activity_type}: {test.name}", description=f"[Aegis] Red Team execution: {test.name}",
) )
logger.info( logger.info(
"Tempo worklog created for test %s by user %s, %ds", "Tempo worklog created for test %s by user %s: %ds on %s",
test.id, getattr(user, "username", user), duration, test.id, getattr(user, "username", user), duration_seconds, work_date,
) )
return result return result
except Exception as e: except Exception as e:
@@ -142,29 +177,3 @@ def auto_log_test_worklog(
test.id, getattr(user, "username", user), e, exc_info=True, test.id, getattr(user, "username", user), e, exc_info=True,
) )
return None return None
def _calculate_duration(test, activity_type: str) -> int:
"""Calculate real duration in seconds from the phase timing fields.
Uses the actual start/end timestamps recorded by the workflow buttons,
so the data cannot be falsified.
"""
from datetime import datetime
now = datetime.utcnow()
if activity_type == "red_team_execution" and test.red_started_at:
delta = now - test.red_started_at
return max(int(delta.total_seconds()), 1)
if activity_type == "blue_team_evaluation" and test.blue_started_at:
delta = now - test.blue_started_at
return max(int(delta.total_seconds()), 1)
# Fallback for legacy activity types
if activity_type == "execution" and test.execution_date and test.created_at:
delta = test.execution_date - test.created_at
return max(int(delta.total_seconds()), 0)
return 0

View File

@@ -327,10 +327,11 @@ def _create_phase_worklog(
test.id, activity_type, duration_seconds, wl.id, test.id, activity_type, duration_seconds, wl.id,
) )
# Sync to Tempo if enabled # Sync to Tempo: only red_team_execution, using the already-computed
# duration so the Tempo entry is identical to the Aegis worklog.
try: try:
from app.services.tempo_service import auto_log_test_worklog from app.services.tempo_service import auto_log_test_worklog
auto_log_test_worklog(db, test, user, activity_type) auto_log_test_worklog(db, test, user, activity_type, duration_seconds)
except Exception as e: except Exception as e:
logger.warning("Tempo sync failed for worklog: %s", e, exc_info=True) logger.warning("Tempo sync failed for worklog: %s", e, exc_info=True)