Get the active ExecutionToken for a mission.
(
mission_id: str,
db: Annotated[AsyncSession, Depends(get_db)],
)
| 568 | description="Get the currently active ExecutionToken for a mission.", |
| 569 | ) |
| 570 | async def get_active_token_by_mission( |
| 571 | mission_id: str, |
| 572 | db: Annotated[AsyncSession, Depends(get_db)], |
| 573 | ) -> ExecutionTokenOut | None: |
| 574 | """Get the active ExecutionToken for a mission.""" |
| 575 | svc = ExecutionTokenService(db) |
| 576 | token = await svc.get_active_token(mission_id) |
| 577 | if token is None: |
| 578 | return None |
| 579 | return ExecutionTokenOut( |
| 580 | token_id=token.token_id, |
| 581 | agent_id=token.agent_id, |
| 582 | mission_id=token.mission_id, |
| 583 | execution_scope=token.execution_scope, |
| 584 | immutable_params=token.immutable_params, |
| 585 | bounded_params=token.bounded_params, |
| 586 | issued_at=token.issued_at, |
| 587 | expires_at=token.expires_at, |
| 588 | boundary_snapshot_id=token.boundary_snapshot_id, |
| 589 | token_signature=token.token_signature, |
| 590 | used=token.used, |
| 591 | is_valid=token.is_valid, |
| 592 | invalidated_at=token.invalidated_at, |
| 593 | invalidation_reason=token.invalidation_reason, |
| 594 | ) |
| 595 | |
| 596 | |
| 597 | @_token.get( |
nothing calls this directly
no test coverage detected