kitos
114 KiB
114 KiB
Campaign Manager Approval Workflow Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Add a manager role that approves campaigns leads build, fixes their launch dates, and reviews justified requests to add/remove tests from an already-active campaign — with every action logged to a queryable timeline.
Architecture: Campaign gains a pending_approval status between draft and active. Leads submit drafts into that queue; a manager (or admin) approves (setting start_date) or rejects (with a mandatory reason, returning it to draft). Once active, leads can no longer edit the test list directly — they file a CampaignModificationRequest (with mandatory justification) that a manager must approve before the underlying test is actually added/removed. All of it reuses the existing generic AuditLog (new action names) and exposes a GET /campaigns/{id}/timeline endpoint, mirroring the one that already exists for tests.
Tech Stack: FastAPI + SQLAlchemy + Alembic (backend), React 19 + TypeScript + TanStack Query (frontend), pytest + in-memory SQLite (tests).
Context for the implementing engineer
- Read backend/app/services/campaign_crud_service.py fully before starting — nearly every backend task edits this file.
- Read backend/app/routers/campaigns.py fully before starting Task 9 onward.
require_any_role(*roles)(inapp/dependencies/auth.py) already letsadminthrough unconditionally — sorequire_any_role("manager")alone is enough to satisfy "admin can also approve." You never need to writerequire_any_role("manager", "admin").- Domain errors (
EntityNotFoundError,BusinessRuleViolation,PermissionViolation) are raised in the service layer and converted to HTTP 404/400/403 automatically byapp/middleware/error_handler.py— routers never raiseHTTPExceptionfor these cases, they just let the exception propagate. - Service functions never call
db.commit()— onlydb.flush(). The router wraps calls inwith UnitOfWork(db) as uow: ... uow.commit(). Follow this exactly. - Backend tests use the fixtures in
backend/tests/conftest.py(client,db,admin_user/admin_token,red_lead_user/red_lead_headers, etc.) against a real (in-memory SQLite) FastAPI app — no mocking. Run pytest from thebackend/directory. - Frontend has no test runner configured. Frontend tasks are verified with
npm run build(TypeScript check) insidefrontend/and manual verification in the browser via the dev server. - This plan deliberately leaves
app/services/campaign_scheduler_service.py(recurring campaign auto-clone, which setsstatus="active"directly) untouched — a recurring child campaign is a re-run of an already-approved template, not a new campaign a lead is proposing, so it doesn't need to re-enter the approval queue.
Task 1: Domain entity — add pending_approval state and transition methods
Files:
-
Modify:
backend/app/domain/entities/campaign.py -
Test:
backend/tests/test_campaign_entity.py -
Step 1: Write the failing tests
Append to backend/tests/test_campaign_entity.py:
# ── 9. Submit for approval ────────────────────────────────────────────
def test_submit_for_approval_from_draft_with_tests_success():
e = _entity("draft", test_count=1)
e.submit_for_approval()
assert e.status == CampaignStatus.pending_approval
def test_submit_for_approval_from_draft_with_zero_tests_raises():
e = _entity("draft", test_count=0)
with pytest.raises(BusinessRuleViolation, match="at least one test"):
e.submit_for_approval()
assert e.status == CampaignStatus.draft
def test_submit_for_approval_from_active_raises():
e = _entity("active", test_count=1)
with pytest.raises(InvalidStateTransition):
e.submit_for_approval()
# ── 10. Approve ────────────────────────────────────────────────────────
def test_approve_from_pending_approval_success():
e = _entity("pending_approval", test_count=1)
e.approve()
assert e.status == CampaignStatus.active
def test_approve_from_draft_raises():
e = _entity("draft", test_count=1)
with pytest.raises(InvalidStateTransition):
e.approve()
# ── 11. Reject ─────────────────────────────────────────────────────────
def test_reject_from_pending_approval_returns_to_draft():
e = _entity("pending_approval", test_count=1)
e.reject()
assert e.status == CampaignStatus.draft
def test_reject_from_draft_raises():
e = _entity("draft", test_count=1)
with pytest.raises(InvalidStateTransition):
e.reject()
- Step 2: Run tests to verify they fail
Run (from backend/): pytest tests/test_campaign_entity.py -v -k "submit_for_approval or test_approve or test_reject"
Expected: FAIL — AttributeError: 'CampaignEntity' object has no attribute 'submit_for_approval' (and CampaignStatus.pending_approval doesn't exist yet).
- Step 3: Implement the state and methods
In backend/app/domain/entities/campaign.py, update CampaignStatus:
class CampaignStatus(str, enum.Enum):
"""Lifecycle states for a campaign."""
draft = "draft"
pending_approval = "pending_approval"
active = "active"
completed = "completed"
archived = "archived"
Update VALID_TRANSITIONS:
VALID_TRANSITIONS: dict[CampaignStatus, list[CampaignStatus]] = {
CampaignStatus.draft: [CampaignStatus.pending_approval],
CampaignStatus.pending_approval: [CampaignStatus.active, CampaignStatus.draft],
CampaignStatus.active: [CampaignStatus.completed],
CampaignStatus.completed: [CampaignStatus.archived],
CampaignStatus.archived: [],
}
Replace the activate method and add submit_for_approval, approve, reject right after it:
def submit_for_approval(self) -> None:
"""Transition the campaign from ``draft`` to ``pending_approval``.
Returns:
None
"""
if not self.can_transition_to(CampaignStatus.pending_approval):
raise InvalidStateTransition(
self.status.value, CampaignStatus.pending_approval.value,
[s.value for s in VALID_TRANSITIONS[self.status]],
)
if self.test_count == 0:
raise BusinessRuleViolation(
"Campaign must have at least one test to submit for approval"
)
self.status = CampaignStatus.pending_approval
def approve(self) -> None:
"""Transition the campaign from ``pending_approval`` to ``active``.
Returns:
None
"""
if not self.can_transition_to(CampaignStatus.active):
raise InvalidStateTransition(
self.status.value, CampaignStatus.active.value,
[s.value for s in VALID_TRANSITIONS[self.status]],
)
self.status = CampaignStatus.active
def reject(self) -> None:
"""Transition the campaign from ``pending_approval`` back to ``draft``.
Returns:
None
"""
if not self.can_transition_to(CampaignStatus.draft):
raise InvalidStateTransition(
self.status.value, CampaignStatus.draft.value,
[s.value for s in VALID_TRANSITIONS[self.status]],
)
self.status = CampaignStatus.draft
Leave the existing activate method as-is (it's superseded by submit_for_approval/approve but removing it would break test_activate_from_draft_with_tests_success etc. — out of scope to touch those).
Update ensure_modifiable so campaigns can still be read while pending approval isn't treated as an error state for callers that only check draft/active — leave this method unchanged; it already raises for anything other than draft/active, which is correct (pending_approval campaigns are frozen).
- Step 4: Run tests to verify they pass
Run: pytest tests/test_campaign_entity.py -v
Expected: PASS — all tests including the pre-existing ones (the old activate/ensure_modifiable tests must still pass unchanged).
- Step 5: Commit
git add backend/app/domain/entities/campaign.py backend/tests/test_campaign_entity.py
git commit -m "feat(campaigns): add pending_approval state to campaign domain entity"
Task 2: ORM models — Campaign approval fields + CampaignModificationRequest
Files:
-
Modify:
backend/app/models/campaign.py -
Modify:
backend/app/models/__init__.py -
Step 1: Add approval fields to
Campaign
In backend/app/models/campaign.py, add these columns right after start_date/scheduled_at (after line 77):
approved_by = Column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="SET NULL"),
nullable=True,
)
approved_at = Column(DateTime, nullable=True)
rejection_reason = Column(Text, nullable=True)
- Step 2: Add the
CampaignModificationRequestmodel
Append to the bottom of backend/app/models/campaign.py, after the CampaignTest class:
class CampaignModificationRequest(Base):
"""A lead's request to add or remove a test from an already-active campaign.
The underlying ``CampaignTest`` row is only created/deleted once a
manager approves the request — see ``approve_modification_request``
in ``campaign_crud_service.py``.
"""
__tablename__ = "campaign_modification_requests"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
campaign_id = Column(
UUID(as_uuid=True),
ForeignKey("campaigns.id", ondelete="CASCADE"),
nullable=False,
)
requested_by = Column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="SET NULL"),
nullable=True,
)
action = Column(String, nullable=False) # add_test, remove_test
test_id = Column(
UUID(as_uuid=True),
ForeignKey("tests.id", ondelete="CASCADE"),
nullable=False,
)
order_index = Column(Integer, nullable=True)
phase = Column(String, nullable=True)
justification = Column(Text, nullable=False)
status = Column(String, nullable=False, default="pending") # pending, approved, rejected
reviewed_by = Column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="SET NULL"),
nullable=True,
)
reviewed_at = Column(DateTime, nullable=True)
review_notes = Column(Text, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
campaign = relationship("Campaign")
test = relationship("Test")
requester = relationship("User", foreign_keys=[requested_by])
reviewer = relationship("User", foreign_keys=[reviewed_by])
__table_args__ = (
Index("ix_campaign_mod_requests_campaign", "campaign_id"),
Index("ix_campaign_mod_requests_status", "status"),
)
- Step 3: Register the new model
In backend/app/models/__init__.py, change line 11:
from app.models.campaign import Campaign, CampaignTest, CampaignModificationRequest
And in __all__ (line 62), change:
"Campaign", "CampaignTest", "CampaignModificationRequest",
- Step 4: Verify the app still imports cleanly
Run (from backend/): python -c "import app.models; print('ok')"
Expected: prints ok with no traceback.
- Step 5: Commit
git add backend/app/models/campaign.py backend/app/models/__init__.py
git commit -m "feat(campaigns): add approval fields and CampaignModificationRequest model"
Task 3: Migration — campaign approval columns
Files:
-
Create:
backend/alembic/versions/b054_add_campaign_approval_fields.py -
Step 1: Write the migration
"""Add approved_by, approved_at, rejection_reason to campaigns.
Revision ID: b054
Revises: b053
Create Date: 2026-07-02
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision = "b054"
down_revision = "b053"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"campaigns",
sa.Column("approved_by", postgresql.UUID(as_uuid=True), nullable=True),
)
op.add_column("campaigns", sa.Column("approved_at", sa.DateTime(), nullable=True))
op.add_column("campaigns", sa.Column("rejection_reason", sa.Text(), nullable=True))
op.create_foreign_key(
"fk_campaigns_approved_by_users",
"campaigns", "users",
["approved_by"], ["id"],
ondelete="SET NULL",
)
def downgrade() -> None:
op.drop_constraint("fk_campaigns_approved_by_users", "campaigns", type_="foreignkey")
op.drop_column("campaigns", "rejection_reason")
op.drop_column("campaigns", "approved_at")
op.drop_column("campaigns", "approved_by")
- Step 2: Verify migration chain is valid
Run (from backend/): alembic heads
Expected: prints b054 (head) — confirming no branch conflicts.
- Step 3: Commit
git add backend/alembic/versions/b054_add_campaign_approval_fields.py
git commit -m "feat(campaigns): migration for campaign approval fields"
Task 4: Migration — campaign_modification_requests table
Files:
-
Create:
backend/alembic/versions/b055_add_campaign_modification_requests.py -
Step 1: Write the migration
"""Add campaign_modification_requests table.
Revision ID: b055
Revises: b054
Create Date: 2026-07-02
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision = "b055"
down_revision = "b054"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"campaign_modification_requests",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"campaign_id", postgresql.UUID(as_uuid=True),
sa.ForeignKey("campaigns.id", ondelete="CASCADE"), nullable=False,
),
sa.Column(
"requested_by", postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True,
),
sa.Column("action", sa.String(), nullable=False),
sa.Column(
"test_id", postgresql.UUID(as_uuid=True),
sa.ForeignKey("tests.id", ondelete="CASCADE"), nullable=False,
),
sa.Column("order_index", sa.Integer(), nullable=True),
sa.Column("phase", sa.String(), nullable=True),
sa.Column("justification", sa.Text(), nullable=False),
sa.Column("status", sa.String(), nullable=False, server_default="pending"),
sa.Column(
"reviewed_by", postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True,
),
sa.Column("reviewed_at", sa.DateTime(), nullable=True),
sa.Column("review_notes", sa.Text(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
op.create_index(
"ix_campaign_mod_requests_campaign",
"campaign_modification_requests", ["campaign_id"],
)
op.create_index(
"ix_campaign_mod_requests_status",
"campaign_modification_requests", ["status"],
)
def downgrade() -> None:
op.drop_index("ix_campaign_mod_requests_status", table_name="campaign_modification_requests")
op.drop_index("ix_campaign_mod_requests_campaign", table_name="campaign_modification_requests")
op.drop_table("campaign_modification_requests")
- Step 2: Verify migration chain is valid
Run (from backend/): alembic heads
Expected: prints b055 (head).
- Step 3: Commit
git add backend/alembic/versions/b055_add_campaign_modification_requests.py
git commit -m "feat(campaigns): migration for campaign_modification_requests table"
Task 5: Test fixtures — manager role
Files:
-
Modify:
backend/tests/conftest.py -
Step 1: Add the
manager_userfixture
Insert right after the blue_lead_user fixture (after line 217) in backend/tests/conftest.py:
@pytest.fixture(scope="function")
def manager_user(db):
"""Create a manager user for testing."""
user = User(
username="manager",
email="manager@test.com",
hashed_password=hash_password("manager123"),
role="manager",
is_active=True,
must_change_password=False,
)
db.add(user)
db.commit()
db.refresh(user)
return user
- Step 2: Add the
manager_tokenandmanager_headersfixtures
Insert right after the blue_lead_headers fixture (after line 297):
@pytest.fixture(scope="function")
def manager_token(client, manager_user):
"""Get an auth token for the manager user."""
response = client.post(
"/api/v1/auth/login",
data={"username": "manager", "password": "manager123"},
)
return response.json()["access_token"]
@pytest.fixture(scope="function")
def manager_headers(manager_token):
"""Return authorization headers for manager user."""
return {"Authorization": f"Bearer {manager_token}"}
- Step 3: Verify the fixtures load without error
Run (from backend/): pytest tests/test_user_api_validation.py -v
Expected: PASS (unrelated test file — this just confirms conftest.py still parses and collects cleanly after the edit).
- Step 4: Commit
git add backend/tests/conftest.py
git commit -m "test(campaigns): add manager role fixtures"
Task 6: Service layer — submit / approve / reject campaign
Files:
-
Modify:
backend/app/services/campaign_crud_service.py -
Test:
backend/tests/test_campaign_approval.py(new file) -
Step 1: Write the failing tests
Create backend/tests/test_campaign_approval.py:
"""Tests for the campaign manager-approval workflow."""
import pytest
from app.models.campaign import Campaign, CampaignTest
from app.models.technique import Technique
from app.models.test import Test
from app.models.enums import TestState
from app.domain.errors import BusinessRuleViolation, EntityNotFoundError, PermissionViolation
from app.services.campaign_crud_service import (
create_campaign,
submit_campaign_for_approval,
approve_campaign,
reject_campaign,
)
@pytest.fixture
def draft_campaign_with_test(db, admin_user):
"""A draft campaign with exactly one test, created by admin_user."""
tech = Technique(mitre_id="T1059", name="Command Line", tactic="execution", platforms=["windows"])
db.add(tech)
db.flush()
campaign = Campaign(name="Approval Test Campaign", type="custom", status="draft", created_by=admin_user.id)
db.add(campaign)
db.flush()
test = Test(technique_id=tech.id, name="T1059 test", state=TestState.draft, created_by=admin_user.id)
db.add(test)
db.flush()
ct = CampaignTest(campaign_id=campaign.id, test_id=test.id, order_index=0)
db.add(ct)
db.commit()
db.refresh(campaign)
return campaign
class TestSubmitCampaignForApproval:
def test_submit_moves_draft_to_pending_approval(self, db, draft_campaign_with_test, admin_user):
campaign = submit_campaign_for_approval(
db, str(draft_campaign_with_test.id),
submitter_id=admin_user.id, submitter_role="admin",
)
assert campaign.status == "pending_approval"
def test_submit_without_tests_raises(self, db, admin_user):
campaign = Campaign(name="Empty", type="custom", status="draft", created_by=admin_user.id)
db.add(campaign)
db.commit()
with pytest.raises(BusinessRuleViolation, match="at least one test"):
submit_campaign_for_approval(
db, str(campaign.id), submitter_id=admin_user.id, submitter_role="admin",
)
def test_submit_by_non_creator_non_admin_raises(self, db, draft_campaign_with_test, red_lead_user):
with pytest.raises(PermissionViolation):
submit_campaign_for_approval(
db, str(draft_campaign_with_test.id),
submitter_id=red_lead_user.id, submitter_role="red_lead",
)
def test_submit_non_draft_raises(self, db, draft_campaign_with_test, admin_user):
draft_campaign_with_test.status = "active"
db.commit()
with pytest.raises(BusinessRuleViolation, match="Only draft campaigns"):
submit_campaign_for_approval(
db, str(draft_campaign_with_test.id),
submitter_id=admin_user.id, submitter_role="admin",
)
class TestApproveCampaign:
def test_approve_sets_start_date_and_activates(self, db, draft_campaign_with_test, admin_user, manager_user=None):
submit_campaign_for_approval(
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
)
db.commit()
campaign = approve_campaign(
db, str(draft_campaign_with_test.id),
approver_id=admin_user.id, start_date="2026-08-01T00:00:00",
)
assert campaign.status == "active"
assert campaign.start_date is not None
assert campaign.approved_by == admin_user.id
assert campaign.approved_at is not None
def test_approve_without_start_date_raises(self, db, draft_campaign_with_test, admin_user):
submit_campaign_for_approval(
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
)
db.commit()
with pytest.raises(BusinessRuleViolation, match="start_date is required"):
approve_campaign(db, str(draft_campaign_with_test.id), approver_id=admin_user.id, start_date="")
def test_approve_non_pending_raises(self, db, draft_campaign_with_test, admin_user):
with pytest.raises(BusinessRuleViolation, match="pending approval"):
approve_campaign(
db, str(draft_campaign_with_test.id),
approver_id=admin_user.id, start_date="2026-08-01T00:00:00",
)
class TestRejectCampaign:
def test_reject_returns_to_draft_with_reason(self, db, draft_campaign_with_test, admin_user):
submit_campaign_for_approval(
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
)
db.commit()
campaign = reject_campaign(
db, str(draft_campaign_with_test.id),
rejecter_id=admin_user.id, reason="Missing scope details",
)
assert campaign.status == "draft"
assert campaign.rejection_reason == "Missing scope details"
def test_reject_without_reason_raises(self, db, draft_campaign_with_test, admin_user):
submit_campaign_for_approval(
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
)
db.commit()
with pytest.raises(BusinessRuleViolation, match="reason is required"):
reject_campaign(db, str(draft_campaign_with_test.id), rejecter_id=admin_user.id, reason="")
def test_reject_non_pending_raises(self, db, draft_campaign_with_test, admin_user):
with pytest.raises(BusinessRuleViolation, match="pending approval"):
reject_campaign(db, str(draft_campaign_with_test.id), rejecter_id=admin_user.id, reason="No")
- Step 2: Run tests to verify they fail
Run (from backend/): pytest tests/test_campaign_approval.py -v
Expected: FAIL — ImportError: cannot import name 'submit_campaign_for_approval' from 'app.services.campaign_crud_service'.
- Step 3: Implement the service functions
In backend/app/services/campaign_crud_service.py, add right after create_campaign (after line 306, before get_campaign_detail):
def submit_campaign_for_approval(
db: Session,
campaign_id: str,
*,
submitter_id: uuid.UUID,
submitter_role: str,
) -> Campaign:
"""Submit a draft campaign into the manager's approval queue.
Raises EntityNotFoundError, PermissionViolation, BusinessRuleViolation.
Does not commit; caller commits.
"""
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
if not campaign:
raise EntityNotFoundError("Campaign", campaign_id)
if str(campaign.created_by) != str(submitter_id) and submitter_role != "admin":
raise PermissionViolation("Only the creator or admin can submit this campaign for approval")
if campaign.status != "draft":
raise BusinessRuleViolation("Only draft campaigns can be submitted for approval")
test_count = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign_id).count()
if test_count == 0:
raise BusinessRuleViolation("Campaign must have at least one test to submit for approval")
campaign.status = "pending_approval"
db.flush()
return campaign
def approve_campaign(
db: Session,
campaign_id: str,
*,
approver_id: uuid.UUID,
start_date: str,
) -> Campaign:
"""Approve a pending campaign: fix its start date and activate it.
Raises EntityNotFoundError, BusinessRuleViolation.
Does not commit; caller commits.
"""
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
if not campaign:
raise EntityNotFoundError("Campaign", campaign_id)
if campaign.status != "pending_approval":
raise BusinessRuleViolation("Only campaigns pending approval can be approved")
if not start_date:
raise BusinessRuleViolation("start_date is required to approve a campaign")
campaign.start_date = datetime.fromisoformat(start_date)
campaign.status = "active"
campaign.approved_by = approver_id
campaign.approved_at = datetime.utcnow()
campaign.rejection_reason = None
db.flush()
return campaign
def reject_campaign(
db: Session,
campaign_id: str,
*,
rejecter_id: uuid.UUID,
reason: str,
) -> Campaign:
"""Reject a pending campaign, returning it to draft with a mandatory reason.
Raises EntityNotFoundError, BusinessRuleViolation.
Does not commit; caller commits.
"""
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
if not campaign:
raise EntityNotFoundError("Campaign", campaign_id)
if campaign.status != "pending_approval":
raise BusinessRuleViolation("Only campaigns pending approval can be rejected")
if not reason or not reason.strip():
raise BusinessRuleViolation("A rejection reason is required")
campaign.status = "draft"
campaign.rejection_reason = reason.strip()
campaign.approved_by = None
campaign.approved_at = None
db.flush()
return campaign
Note: rejecter_id is accepted for symmetry with approve_campaign's approver_id but isn't stored (there's no rejected_by column — rejection_reason plus the audit log entry the router writes are the record of who rejected). Leave the parameter in the signature since the router will always pass it and a future column could use it.
- Step 4: Run tests to verify they pass
Run: pytest tests/test_campaign_approval.py -v
Expected: PASS — all 10 tests green.
- Step 5: Commit
git add backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval.py
git commit -m "feat(campaigns): submit/approve/reject service functions"
Task 7: Service layer — gate test add/remove to draft, add modification requests
Files:
-
Modify:
backend/app/services/campaign_crud_service.py -
Test:
backend/tests/test_campaign_approval.py -
Step 1: Write the failing tests
Append to backend/tests/test_campaign_approval.py:
from app.services.campaign_crud_service import (
add_test_to_campaign,
remove_test_from_campaign,
create_modification_request,
approve_modification_request,
reject_modification_request,
list_modification_requests,
)
@pytest.fixture
def active_campaign_with_test(db, draft_campaign_with_test, admin_user):
"""An active campaign (bypasses the approval flow directly for test setup)."""
draft_campaign_with_test.status = "active"
db.commit()
db.refresh(draft_campaign_with_test)
return draft_campaign_with_test
class TestDirectTestEditingGate:
def test_add_test_blocked_on_active_campaign(self, db, active_campaign_with_test, admin_user):
tech = Technique(mitre_id="T1078", name="Valid Accounts", tactic="persistence", platforms=["windows"])
db.add(tech)
db.flush()
new_test = Test(technique_id=tech.id, name="New test", state=TestState.draft, created_by=admin_user.id)
db.add(new_test)
db.commit()
with pytest.raises(BusinessRuleViolation, match="draft"):
add_test_to_campaign(db, str(active_campaign_with_test.id), test_id=str(new_test.id))
def test_remove_test_blocked_on_active_campaign(self, db, active_campaign_with_test):
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
with pytest.raises(BusinessRuleViolation, match="draft"):
remove_test_from_campaign(db, str(active_campaign_with_test.id), str(ct.id))
class TestModificationRequests:
def test_create_add_test_request(self, db, active_campaign_with_test, admin_user):
tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"])
db.add(tech)
db.flush()
new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=admin_user.id)
db.add(new_test)
db.commit()
request = create_modification_request(
db, str(active_campaign_with_test.id),
requester_id=admin_user.id, action="add_test",
test_id=str(new_test.id), justification="Coverage gap found in review",
)
assert request.status == "pending"
# Underlying campaign is untouched until approval
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
assert len(cts) == 1
def test_create_request_without_justification_raises(self, db, active_campaign_with_test, admin_user):
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
with pytest.raises(BusinessRuleViolation, match="justification is required"):
create_modification_request(
db, str(active_campaign_with_test.id),
requester_id=admin_user.id, action="remove_test",
test_id=str(ct.test_id), justification="",
)
def test_create_request_on_non_active_campaign_raises(self, db, draft_campaign_with_test, admin_user):
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == draft_campaign_with_test.id).first()
with pytest.raises(BusinessRuleViolation, match="active campaigns"):
create_modification_request(
db, str(draft_campaign_with_test.id),
requester_id=admin_user.id, action="remove_test",
test_id=str(ct.test_id), justification="Not needed",
)
def test_approve_add_test_request_applies_change(self, db, active_campaign_with_test, admin_user):
tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"])
db.add(tech)
db.flush()
new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=admin_user.id)
db.add(new_test)
db.commit()
request = create_modification_request(
db, str(active_campaign_with_test.id),
requester_id=admin_user.id, action="add_test",
test_id=str(new_test.id), justification="Coverage gap found in review",
)
db.commit()
approved = approve_modification_request(db, str(request.id), reviewer_id=admin_user.id)
assert approved.status == "approved"
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
assert len(cts) == 2
assert any(ct.test_id == new_test.id for ct in cts)
def test_approve_remove_test_request_applies_change(self, db, active_campaign_with_test, admin_user):
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
request = create_modification_request(
db, str(active_campaign_with_test.id),
requester_id=admin_user.id, action="remove_test",
test_id=str(ct.test_id), justification="Test superseded",
)
db.commit()
approve_modification_request(db, str(request.id), reviewer_id=admin_user.id)
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
assert len(cts) == 0
def test_reject_request_leaves_campaign_untouched(self, db, active_campaign_with_test, admin_user):
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
request = create_modification_request(
db, str(active_campaign_with_test.id),
requester_id=admin_user.id, action="remove_test",
test_id=str(ct.test_id), justification="Test superseded",
)
db.commit()
rejected = reject_modification_request(
db, str(request.id), reviewer_id=admin_user.id, review_notes="Still needed for coverage",
)
assert rejected.status == "rejected"
assert rejected.review_notes == "Still needed for coverage"
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
assert len(cts) == 1
def test_reject_without_notes_raises(self, db, active_campaign_with_test, admin_user):
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
request = create_modification_request(
db, str(active_campaign_with_test.id),
requester_id=admin_user.id, action="remove_test",
test_id=str(ct.test_id), justification="Test superseded",
)
db.commit()
with pytest.raises(BusinessRuleViolation, match="Review notes are required"):
reject_modification_request(db, str(request.id), reviewer_id=admin_user.id, review_notes="")
def test_list_modification_requests_filters_by_status(self, db, active_campaign_with_test, admin_user):
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
create_modification_request(
db, str(active_campaign_with_test.id),
requester_id=admin_user.id, action="remove_test",
test_id=str(ct.test_id), justification="Test superseded",
)
db.commit()
pending = list_modification_requests(db, status="pending")
assert len(pending) == 1
assert pending[0]["action"] == "remove_test"
- Step 2: Run tests to verify they fail
Run: pytest tests/test_campaign_approval.py -v -k Modification or DirectTestEditingGate
Expected: FAIL — ImportError for the new functions, and the existing add_test_to_campaign/remove_test_from_campaign gate tests fail because those functions currently accept active campaigns too.
- Step 3: Tighten the status gate on
add_test_to_campaignandremove_test_from_campaign
In backend/app/services/campaign_crud_service.py, change the add_test_to_campaign signature (around line 379) to accept an allowed_statuses keyword, defaulting to draft-only:
def add_test_to_campaign(
db: Session,
campaign_id: str,
*,
test_id: str,
order_index: Optional[int] = None,
depends_on: Optional[str] = None,
phase: Optional[str] = None,
allowed_statuses: tuple[str, ...] = ("draft",),
) -> dict:
And change its status check (currently if campaign.status not in ("draft", "active"):) to:
if campaign.status not in allowed_statuses:
raise BusinessRuleViolation(
f"Can only add tests to a campaign in one of: {', '.join(allowed_statuses)}"
)
Do the same for remove_test_from_campaign (around line 495): add allowed_statuses: tuple[str, ...] = ("draft",) to its signature, and change its check to:
if campaign.status not in allowed_statuses:
raise BusinessRuleViolation(
f"Can only modify tests on a campaign in one of: {', '.join(allowed_statuses)}"
)
This is the only change to either function's body — everything else (order_index calculation, circular-dependency validation, phase inference, dependent-cleanup) stays exactly as-is.
- Step 4: Add the modification-request functions
Add the import for the new model at the top of backend/app/services/campaign_crud_service.py (change the existing campaign model import line):
from app.models.campaign import Campaign, CampaignModificationRequest, CampaignTest
Append these functions at the end of the file:
# ── Campaign Modification Requests ──────────────────────────────────────
def serialize_modification_request(db: Session, request: CampaignModificationRequest) -> dict:
"""Serialize a modification request, including the test's display name."""
test = request.test
return {
"id": str(request.id),
"campaign_id": str(request.campaign_id),
"campaign_name": request.campaign.name if request.campaign else None,
"requested_by": str(request.requested_by) if request.requested_by else None,
"action": request.action,
"test_id": str(request.test_id),
"test_name": test.name if test else None,
"order_index": request.order_index,
"phase": request.phase,
"justification": request.justification,
"status": request.status,
"reviewed_by": str(request.reviewed_by) if request.reviewed_by else None,
"reviewed_at": request.reviewed_at.isoformat() if request.reviewed_at else None,
"review_notes": request.review_notes,
"created_at": request.created_at.isoformat() if request.created_at else None,
}
def create_modification_request(
db: Session,
campaign_id: str,
*,
requester_id: uuid.UUID,
action: str,
test_id: str,
justification: str,
order_index: Optional[int] = None,
phase: Optional[str] = None,
) -> CampaignModificationRequest:
"""File a request to add/remove a test on an active campaign.
Raises EntityNotFoundError, BusinessRuleViolation.
Does not commit; caller commits.
"""
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
if not campaign:
raise EntityNotFoundError("Campaign", campaign_id)
if campaign.status != "active":
raise BusinessRuleViolation("Modification requests can only be created for active campaigns")
if action not in ("add_test", "remove_test"):
raise BusinessRuleViolation("action must be 'add_test' or 'remove_test'")
if not justification or not justification.strip():
raise BusinessRuleViolation("A justification is required")
test = db.query(Test).filter(Test.id == test_id).first()
if not test:
raise EntityNotFoundError("Test", test_id)
if action == "remove_test":
ct = (
db.query(CampaignTest)
.filter(CampaignTest.campaign_id == campaign_id, CampaignTest.test_id == test_id)
.first()
)
if not ct:
raise EntityNotFoundError("CampaignTest", test_id)
request = CampaignModificationRequest(
campaign_id=campaign_id,
requested_by=requester_id,
action=action,
test_id=test_id,
order_index=order_index,
phase=phase,
justification=justification.strip(),
status="pending",
)
db.add(request)
db.flush()
return request
def approve_modification_request(
db: Session,
request_id: str,
*,
reviewer_id: uuid.UUID,
) -> CampaignModificationRequest:
"""Approve a pending modification request and apply the underlying test change.
Raises EntityNotFoundError, BusinessRuleViolation.
Does not commit; caller commits.
"""
request = (
db.query(CampaignModificationRequest)
.filter(CampaignModificationRequest.id == request_id)
.first()
)
if not request:
raise EntityNotFoundError("CampaignModificationRequest", request_id)
if request.status != "pending":
raise BusinessRuleViolation("Only pending modification requests can be approved")
if request.action == "add_test":
add_test_to_campaign(
db, str(request.campaign_id),
test_id=str(request.test_id),
order_index=request.order_index,
phase=request.phase,
allowed_statuses=("active",),
)
else:
ct = (
db.query(CampaignTest)
.filter(
CampaignTest.campaign_id == request.campaign_id,
CampaignTest.test_id == request.test_id,
)
.first()
)
if not ct:
raise EntityNotFoundError("CampaignTest", str(request.test_id))
remove_test_from_campaign(
db, str(request.campaign_id), str(ct.id), allowed_statuses=("active",),
)
request.status = "approved"
request.reviewed_by = reviewer_id
request.reviewed_at = datetime.utcnow()
db.flush()
return request
def reject_modification_request(
db: Session,
request_id: str,
*,
reviewer_id: uuid.UUID,
review_notes: str,
) -> CampaignModificationRequest:
"""Reject a pending modification request. No change is applied to the campaign.
Raises EntityNotFoundError, BusinessRuleViolation.
Does not commit; caller commits.
"""
request = (
db.query(CampaignModificationRequest)
.filter(CampaignModificationRequest.id == request_id)
.first()
)
if not request:
raise EntityNotFoundError("CampaignModificationRequest", request_id)
if request.status != "pending":
raise BusinessRuleViolation("Only pending modification requests can be rejected")
if not review_notes or not review_notes.strip():
raise BusinessRuleViolation("Review notes are required to reject a modification request")
request.status = "rejected"
request.reviewed_by = reviewer_id
request.reviewed_at = datetime.utcnow()
request.review_notes = review_notes.strip()
db.flush()
return request
def list_modification_requests(
db: Session,
*,
campaign_id: Optional[str] = None,
status: Optional[str] = None,
) -> list[dict]:
"""List modification requests, optionally filtered by campaign and/or status."""
query = db.query(CampaignModificationRequest)
if campaign_id:
query = query.filter(CampaignModificationRequest.campaign_id == campaign_id)
if status:
query = query.filter(CampaignModificationRequest.status == status)
requests = query.order_by(CampaignModificationRequest.created_at.desc()).all()
return [serialize_modification_request(db, r) for r in requests]
- Step 5: Run tests to verify they pass
Run: pytest tests/test_campaign_approval.py -v
Expected: PASS — all tests from Task 6 and Task 7 green.
- Step 6: Run the full campaign test suite to check for regressions
Run: pytest tests/test_campaigns_and_snapshots.py -v
Expected: PASS — the existing test_circular_dependency_prevention etc. tests still pass (they operate on draft campaigns, unaffected by the tightened gate).
- Step 7: Commit
git add backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval.py
git commit -m "feat(campaigns): gate test edits to draft, add modification request workflow"
Task 8: Service layer — campaign timeline
Files:
-
Modify:
backend/app/services/campaign_crud_service.py -
Test:
backend/tests/test_campaign_approval.py -
Step 1: Write the failing test
Append to backend/tests/test_campaign_approval.py:
from app.services.audit_service import log_action
from app.services.campaign_crud_service import get_campaign_timeline
class TestCampaignTimeline:
def test_timeline_returns_entries_in_chronological_order(self, db, draft_campaign_with_test, admin_user):
log_action(
db, user_id=admin_user.id, action="create_campaign",
entity_type="campaign", entity_id=str(draft_campaign_with_test.id), details={},
)
log_action(
db, user_id=admin_user.id, action="submit_campaign_for_approval",
entity_type="campaign", entity_id=str(draft_campaign_with_test.id), details={},
)
db.commit()
timeline = get_campaign_timeline(db, str(draft_campaign_with_test.id))
assert len(timeline) == 2
assert timeline[0]["action"] == "create_campaign"
assert timeline[1]["action"] == "submit_campaign_for_approval"
def test_timeline_missing_campaign_raises(self, db):
import uuid as uuid_mod
with pytest.raises(EntityNotFoundError):
get_campaign_timeline(db, str(uuid_mod.uuid4()))
- Step 2: Run test to verify it fails
Run: pytest tests/test_campaign_approval.py -v -k CampaignTimeline
Expected: FAIL — ImportError: cannot import name 'get_campaign_timeline'.
- Step 3: Implement
get_campaign_timeline
Add the import at the top of backend/app/services/campaign_crud_service.py:
from app.models.audit import AuditLog
Append the function at the end of the file:
def get_campaign_timeline(db: Session, campaign_id: str) -> list[dict]:
"""Return the chronological audit-log history for a campaign.
Raises EntityNotFoundError if the campaign does not exist.
"""
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
if not campaign:
raise EntityNotFoundError("Campaign", campaign_id)
logs = (
db.query(AuditLog)
.filter(AuditLog.entity_type == "campaign", AuditLog.entity_id == str(campaign_id))
.order_by(AuditLog.timestamp.asc())
.all()
)
return [
{
"id": str(log.id),
"action": log.action,
"user_id": str(log.user_id) if log.user_id else None,
"timestamp": log.timestamp.isoformat() if log.timestamp else None,
"details": log.details or {},
}
for log in logs
]
- Step 4: Run test to verify it passes
Run: pytest tests/test_campaign_approval.py -v
Expected: PASS — full file green.
- Step 5: Commit
git add backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval.py
git commit -m "feat(campaigns): add get_campaign_timeline service function"
Task 9: Router — submit / approve / reject endpoints
Files:
-
Modify:
backend/app/routers/campaigns.py -
Test:
backend/tests/test_campaign_approval_router.py(new file) -
Step 1: Write the failing tests
Create backend/tests/test_campaign_approval_router.py:
"""Router-level tests for the campaign manager-approval workflow."""
from app.models.campaign import Campaign, CampaignTest
from app.models.technique import Technique
from app.models.test import Test
from app.models.enums import TestState
def _make_draft_campaign(db, owner_id):
tech = Technique(mitre_id="T1059", name="Command Line", tactic="execution", platforms=["windows"])
db.add(tech)
db.flush()
campaign = Campaign(name="Router Test Campaign", type="custom", status="draft", created_by=owner_id)
db.add(campaign)
db.flush()
test = Test(technique_id=tech.id, name="T1059 test", state=TestState.draft, created_by=owner_id)
db.add(test)
db.flush()
db.add(CampaignTest(campaign_id=campaign.id, test_id=test.id, order_index=0))
db.commit()
db.refresh(campaign)
return campaign
def test_lead_can_submit_own_campaign(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
resp = client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers)
assert resp.status_code == 200
assert resp.json()["status"] == "pending_approval"
def test_red_tech_cannot_submit(client, db, red_lead_user, red_tech_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
resp = client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_tech_headers)
assert resp.status_code == 403
def test_manager_can_approve_and_sets_start_date(client, db, red_lead_user, red_lead_headers, manager_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers)
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/approve",
json={"start_date": "2026-09-01T00:00:00"},
headers=manager_headers,
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "active"
assert body["start_date"] is not None
def test_lead_cannot_approve(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers)
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/approve",
json={"start_date": "2026-09-01T00:00:00"},
headers=red_lead_headers,
)
assert resp.status_code == 403
def test_admin_can_approve_as_manager_backup(client, db, red_lead_user, red_lead_headers, auth_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers)
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/approve",
json={"start_date": "2026-09-01T00:00:00"},
headers=auth_headers,
)
assert resp.status_code == 200
def test_manager_can_reject_with_reason(client, db, red_lead_user, red_lead_headers, manager_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers)
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/reject",
json={"reason": "Needs more detail"},
headers=manager_headers,
)
assert resp.status_code == 200
assert resp.json()["status"] == "draft"
def test_manager_reject_without_reason_rejected_by_validation(client, db, red_lead_user, red_lead_headers, manager_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers)
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/reject",
json={"reason": ""},
headers=manager_headers,
)
assert resp.status_code == 400
- Step 2: Run tests to verify they fail
Run: pytest tests/test_campaign_approval_router.py -v
Expected: FAIL — 404 Not Found for /submit, /approve, /reject (routes don't exist yet).
- Step 3: Add the Pydantic payloads and endpoints
In backend/app/routers/campaigns.py, add the import for the new service functions (extend the existing multi-import block for campaign_crud_service — add right after the activate_campaign as crud_activate import around line 92):
from app.services.campaign_crud_service import (
submit_campaign_for_approval as crud_submit,
)
from app.services.campaign_crud_service import (
approve_campaign as crud_approve,
)
from app.services.campaign_crud_service import (
reject_campaign as crud_reject,
)
Add two new Pydantic payloads near the other payload classes (after SchedulePayload, around line 173):
class ApprovePayload(BaseModel):
"""Payload for a manager approving a pending campaign."""
start_date: str # ISO date/datetime — required, campaign won't activate without it
class RejectPayload(BaseModel):
"""Payload for a manager rejecting a pending campaign."""
reason: str
Add the three endpoints right after update_campaign (after line 386, before the DELETE /campaigns/{id} section):
# ---------------------------------------------------------------------------
# POST /campaigns/{id}/submit — Submit draft for manager approval
# ---------------------------------------------------------------------------
@router.post("/{campaign_id}/submit")
def submit_campaign(
campaign_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> dict:
"""Submit a draft campaign into the manager's approval queue."""
with UnitOfWork(db) as uow:
campaign = crud_submit(
db, campaign_id,
submitter_id=current_user.id,
submitter_role=current_user.role,
)
log_action(
db,
user_id=current_user.id,
action="submit_campaign_for_approval",
entity_type="campaign",
entity_id=campaign.id,
details={"name": campaign.name},
)
uow.commit()
db.refresh(campaign)
return serialize_campaign(db, campaign)
# ---------------------------------------------------------------------------
# POST /campaigns/{id}/approve — Manager approves a pending campaign
# ---------------------------------------------------------------------------
@router.post("/{campaign_id}/approve")
def approve_campaign_endpoint(
campaign_id: str,
payload: ApprovePayload,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("manager")),
) -> dict:
"""Manager approves a pending campaign, fixing its start date and activating it."""
with UnitOfWork(db) as uow:
campaign = crud_approve(
db, campaign_id,
approver_id=current_user.id,
start_date=payload.start_date,
)
log_action(
db,
user_id=current_user.id,
action="approve_campaign",
entity_type="campaign",
entity_id=campaign.id,
details={"start_date": payload.start_date},
)
uow.commit()
db.refresh(campaign)
return serialize_campaign(db, campaign)
# ---------------------------------------------------------------------------
# POST /campaigns/{id}/reject — Manager rejects a pending campaign
# ---------------------------------------------------------------------------
@router.post("/{campaign_id}/reject")
def reject_campaign_endpoint(
campaign_id: str,
payload: RejectPayload,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("manager")),
) -> dict:
"""Manager rejects a pending campaign, returning it to draft with a reason."""
with UnitOfWork(db) as uow:
campaign = crud_reject(
db, campaign_id,
rejecter_id=current_user.id,
reason=payload.reason,
)
log_action(
db,
user_id=current_user.id,
action="reject_campaign",
entity_type="campaign",
entity_id=campaign.id,
details={"reason": payload.reason},
)
uow.commit()
db.refresh(campaign)
return serialize_campaign(db, campaign)
- Step 4: Add
approved_by,approved_at,rejection_reasonto the serialized campaign
In serialize_campaign in backend/app/services/campaign_crud_service.py, add these three keys right after "completed_at" (around line 127):
"approved_by": str(campaign.approved_by) if campaign.approved_by else None,
"approved_at": campaign.approved_at.isoformat() if campaign.approved_at else None,
"rejection_reason": campaign.rejection_reason,
- Step 5: Run tests to verify they pass
Run: pytest tests/test_campaign_approval_router.py -v
Expected: PASS — all 7 tests green.
- Step 6: Run the full backend test suite to check for regressions
Run (from backend/): pytest tests/ -v
Expected: PASS — no pre-existing test broken. Pay special attention to test_campaigns_and_snapshots.py.
- Step 7: Commit
git add backend/app/routers/campaigns.py backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval_router.py
git commit -m "feat(campaigns): submit/approve/reject router endpoints"
Task 10: Router — modification request endpoints
Files:
-
Modify:
backend/app/routers/campaigns.py -
Test:
backend/tests/test_campaign_approval_router.py -
Step 1: Write the failing tests
Append to backend/tests/test_campaign_approval_router.py:
def _activate_campaign_directly(db, campaign):
"""Test helper — bypass the approval flow to get straight to 'active' for setup."""
campaign.status = "active"
db.commit()
db.refresh(campaign)
return campaign
def test_lead_can_create_modification_request(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
_activate_campaign_directly(db, campaign)
tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"])
db.add(tech)
db.flush()
new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=red_lead_user.id)
db.add(new_test)
db.commit()
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/modification-requests",
json={"action": "add_test", "test_id": str(new_test.id), "justification": "Coverage gap"},
headers=red_lead_headers,
)
assert resp.status_code == 201
assert resp.json()["status"] == "pending"
def test_modification_request_without_justification_rejected(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
_activate_campaign_directly(db, campaign)
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/modification-requests",
json={"action": "remove_test", "test_id": str(ct.test_id), "justification": ""},
headers=red_lead_headers,
)
assert resp.status_code == 400
def test_manager_can_list_pending_modification_requests(client, db, red_lead_user, red_lead_headers, manager_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
_activate_campaign_directly(db, campaign)
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
client.post(
f"/api/v1/campaigns/{campaign.id}/modification-requests",
json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
headers=red_lead_headers,
)
resp = client.get("/api/v1/campaigns/modification-requests/pending", headers=manager_headers)
assert resp.status_code == 200
assert len(resp.json()) == 1
def test_manager_can_approve_modification_request(client, db, red_lead_user, red_lead_headers, manager_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
_activate_campaign_directly(db, campaign)
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
create_resp = client.post(
f"/api/v1/campaigns/{campaign.id}/modification-requests",
json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
headers=red_lead_headers,
)
request_id = create_resp.json()["id"]
resp = client.post(
f"/api/v1/campaigns/modification-requests/{request_id}/approve",
headers=manager_headers,
)
assert resp.status_code == 200
assert resp.json()["status"] == "approved"
remaining = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).all()
assert len(remaining) == 0
def test_manager_can_reject_modification_request(client, db, red_lead_user, red_lead_headers, manager_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
_activate_campaign_directly(db, campaign)
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
create_resp = client.post(
f"/api/v1/campaigns/{campaign.id}/modification-requests",
json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
headers=red_lead_headers,
)
request_id = create_resp.json()["id"]
resp = client.post(
f"/api/v1/campaigns/modification-requests/{request_id}/reject",
json={"review_notes": "Still needed"},
headers=manager_headers,
)
assert resp.status_code == 200
assert resp.json()["status"] == "rejected"
remaining = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).all()
assert len(remaining) == 1
def test_lead_cannot_approve_modification_request(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
_activate_campaign_directly(db, campaign)
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
create_resp = client.post(
f"/api/v1/campaigns/{campaign.id}/modification-requests",
json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
headers=red_lead_headers,
)
request_id = create_resp.json()["id"]
resp = client.post(
f"/api/v1/campaigns/modification-requests/{request_id}/approve",
headers=red_lead_headers,
)
assert resp.status_code == 403
def test_direct_add_test_blocked_on_active_campaign_via_router(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
_activate_campaign_directly(db, campaign)
tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"])
db.add(tech)
db.flush()
new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=red_lead_user.id)
db.add(new_test)
db.commit()
resp = client.post(
f"/api/v1/campaigns/{campaign.id}/tests",
json={"test_id": str(new_test.id)},
headers=red_lead_headers,
)
assert resp.status_code == 400
- Step 2: Run tests to verify they fail
Run: pytest tests/test_campaign_approval_router.py -v -k modification
Expected: FAIL — 404 Not Found for the new routes (test_direct_add_test_blocked... should already pass since Task 7 tightened the service layer, but the route-level test confirms it end-to-end).
- Step 3: Add the payloads and endpoints
In backend/app/routers/campaigns.py, add the service imports (extend the multi-import block, same place as Task 9's imports):
from app.services.campaign_crud_service import (
create_modification_request as crud_create_mod_request,
)
from app.services.campaign_crud_service import (
approve_modification_request as crud_approve_mod_request,
)
from app.services.campaign_crud_service import (
reject_modification_request as crud_reject_mod_request,
)
from app.services.campaign_crud_service import (
list_modification_requests as crud_list_mod_requests,
)
from app.services.campaign_crud_service import (
serialize_modification_request as crud_serialize_mod_request,
)
Add the payloads near ApprovePayload/RejectPayload:
class ModificationRequestPayload(BaseModel):
"""Payload for a lead requesting to add/remove a test on an active campaign."""
action: str # add_test | remove_test
test_id: str
justification: str
order_index: Optional[int] = None
phase: Optional[str] = None
class RejectModificationPayload(BaseModel):
"""Payload for a manager rejecting a modification request."""
review_notes: str
Add the endpoints right after the reject_campaign_endpoint function from Task 9:
# ---------------------------------------------------------------------------
# POST /campaigns/{id}/modification-requests — Request a test add/remove
# ---------------------------------------------------------------------------
@router.post("/{campaign_id}/modification-requests", status_code=201)
def create_modification_request_endpoint(
campaign_id: str,
payload: ModificationRequestPayload,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> dict:
"""File a request to add/remove a test on an active campaign — needs manager approval."""
with UnitOfWork(db) as uow:
request = crud_create_mod_request(
db, campaign_id,
requester_id=current_user.id,
action=payload.action,
test_id=payload.test_id,
justification=payload.justification,
order_index=payload.order_index,
phase=payload.phase,
)
log_action(
db,
user_id=current_user.id,
action="request_campaign_modification",
entity_type="campaign",
entity_id=campaign_id,
details={
"action": payload.action,
"test_id": payload.test_id,
"justification": payload.justification,
},
)
uow.commit()
return crud_serialize_mod_request(db, request)
# ---------------------------------------------------------------------------
# GET /campaigns/{id}/modification-requests — List requests for one campaign
# ---------------------------------------------------------------------------
@router.get("/{campaign_id}/modification-requests")
def list_campaign_modification_requests_endpoint(
campaign_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> list:
"""List modification requests filed against a specific campaign."""
return crud_list_mod_requests(db, campaign_id=campaign_id)
# ---------------------------------------------------------------------------
# GET /campaigns/modification-requests/pending — Manager's global queue
# ---------------------------------------------------------------------------
@router.get("/modification-requests/pending")
def list_pending_modification_requests_endpoint(
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("manager")),
) -> list:
"""List all modification requests awaiting manager review, across all campaigns."""
return crud_list_mod_requests(db, status="pending")
# ---------------------------------------------------------------------------
# POST /campaigns/modification-requests/{id}/approve — Apply the requested change
# ---------------------------------------------------------------------------
@router.post("/modification-requests/{request_id}/approve")
def approve_modification_request_endpoint(
request_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("manager")),
) -> dict:
"""Manager approves a modification request — the test change is applied now."""
with UnitOfWork(db) as uow:
request = crud_approve_mod_request(db, request_id, reviewer_id=current_user.id)
log_action(
db,
user_id=current_user.id,
action="approve_campaign_modification",
entity_type="campaign",
entity_id=str(request.campaign_id),
details={
"request_id": request_id,
"action": request.action,
"test_id": str(request.test_id),
},
)
uow.commit()
return crud_serialize_mod_request(db, request)
# ---------------------------------------------------------------------------
# POST /campaigns/modification-requests/{id}/reject — Deny the requested change
# ---------------------------------------------------------------------------
@router.post("/modification-requests/{request_id}/reject")
def reject_modification_request_endpoint(
request_id: str,
payload: RejectModificationPayload,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("manager")),
) -> dict:
"""Manager rejects a modification request. No change is applied."""
with UnitOfWork(db) as uow:
request = crud_reject_mod_request(
db, request_id,
reviewer_id=current_user.id,
review_notes=payload.review_notes,
)
log_action(
db,
user_id=current_user.id,
action="reject_campaign_modification",
entity_type="campaign",
entity_id=str(request.campaign_id),
details={"request_id": request_id, "review_notes": payload.review_notes},
)
uow.commit()
return crud_serialize_mod_request(db, request)
Note: GET /campaigns/modification-requests/pending must be declared before GET /campaigns/{campaign_id} would otherwise be — but since that route has a different path segment count (2 segments vs. 1), there's no actual routing conflict regardless of declaration order in this file. Add these new routes in the order shown above for readability.
- Step 4: Run tests to verify they pass
Run: pytest tests/test_campaign_approval_router.py -v
Expected: PASS — all tests green, including test_direct_add_test_blocked_on_active_campaign_via_router.
- Step 5: Commit
git add backend/app/routers/campaigns.py backend/tests/test_campaign_approval_router.py
git commit -m "feat(campaigns): modification-request router endpoints"
Task 11: Router — timeline endpoint, admin-only manual activation, drop lead-set dates
Files:
-
Modify:
backend/app/routers/campaigns.py -
Modify:
backend/app/services/campaign_service.py -
Test:
backend/tests/test_campaign_approval_router.py -
Step 1: Write the failing tests
Append to backend/tests/test_campaign_approval_router.py:
def test_get_campaign_timeline(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
client.post(f"/api/v1/campaigns/{campaign.id}/submit", headers=red_lead_headers)
resp = client.get(f"/api/v1/campaigns/{campaign.id}/timeline", headers=red_lead_headers)
assert resp.status_code == 200
actions = [e["action"] for e in resp.json()]
assert "submit_campaign_for_approval" in actions
def test_lead_cannot_directly_activate_draft_campaign(client, db, red_lead_user, red_lead_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
resp = client.post(f"/api/v1/campaigns/{campaign.id}/activate", headers=red_lead_headers)
assert resp.status_code == 403
def test_admin_can_still_directly_activate_as_override(client, db, red_lead_user, red_lead_headers, auth_headers):
campaign = _make_draft_campaign(db, red_lead_user.id)
resp = client.post(f"/api/v1/campaigns/{campaign.id}/activate", headers=auth_headers)
assert resp.status_code == 200
assert resp.json()["status"] == "active"
def test_create_campaign_payload_has_no_start_date_field(client, db, red_lead_headers):
resp = client.post(
"/api/v1/campaigns",
json={"name": "No date campaign", "start_date": "2026-01-01"},
headers=red_lead_headers,
)
assert resp.status_code == 201
# start_date silently ignored — only the manager can ever set it, via /approve
assert resp.json()["start_date"] is None
- Step 2: Run tests to verify they fail
Run: pytest tests/test_campaign_approval_router.py -v -k "timeline or activate or start_date"
Expected: FAIL — 404 for /timeline, 403-expecting tests get 200 (lead can still activate directly today), and the create-campaign test gets a non-null start_date.
- Step 3: Add the timeline endpoint
Import the new service function in backend/app/routers/campaigns.py (extend the multi-import block):
from app.services.campaign_crud_service import (
get_campaign_timeline as crud_get_timeline,
)
Add the endpoint right after get_campaign_history (after line 856):
# ---------------------------------------------------------------------------
# GET /campaigns/{id}/timeline — Audit-log history for this campaign
# ---------------------------------------------------------------------------
@router.get("/{campaign_id}/timeline")
def get_campaign_timeline_endpoint(
campaign_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> list:
"""Return the chronological audit-log history for a campaign."""
return crud_get_timeline(db, campaign_id)
- Step 4: Restrict direct
/activateto admin only
In backend/app/routers/campaigns.py, change the activate_campaign endpoint's dependency (around line 516) from:
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
to:
current_user: User = Depends(require_any_role("admin")),
This closes the loophole where a lead could bypass the approval queue entirely by calling the old direct-activation endpoint. require_any_role("admin") only ever lets admin through (per its own always-pass rule), so this becomes an admin-only emergency override, matching the "admin also approves" pattern used for /approve and /reject.
- Step 5: Remove
start_datefrom lead-facing create/update/generate payloads
In backend/app/routers/campaigns.py:
- Remove the
start_datefield fromCampaignCreate(delete the linestart_date: Optional[str] = None # ISO date...around line 127). - Remove the
start_datefield fromCampaignUpdate(delete the equivalent line around line 146). - Remove
start_date: Optional[str] = NonefromGenerateFromActorPayload(around line 696) — replace the class body withpass:
class GenerateFromActorPayload(BaseModel):
pass
- In the
generate_campaign_from_actorendpoint, remove the two lines that parse and forwardpayload.start_date:
campaign = generate_campaign_from_threat_actor(
db,
uuid.UUID(actor_id),
current_user,
)
(delete the start_date_parsed = ... line entirely and the start_date=start_date_parsed kwarg).
In backend/app/services/campaign_service.py, remove the start_date parameter from generate_campaign_from_threat_actor's signature and drop start_date=start_date from the Campaign(...) construction inside it (around lines 183 and 241).
In backend/app/services/campaign_crud_service.py, remove the start_date parameter from create_campaign's signature and drop start_date=datetime.fromisoformat(start_date) if start_date else None, from its Campaign(...) construction (around lines 278 and 299). Also remove the start_date handling block from update_campaign (the if "start_date" in fields and fields["start_date"]: block around line 364) — with start_date no longer in the request schema this is dead code, but removing it makes clear the field is intentionally unreachable via this function now.
- Step 6: Run tests to verify they pass
Run: pytest tests/test_campaign_approval_router.py -v
Expected: PASS — all tests in the file green.
- Step 7: Run the full backend test suite
Run (from backend/): pytest tests/ -v
Expected: PASS. If test_campaigns_and_snapshots.py or any other file passes start_date to create_campaign/generate_campaign_from_threat_actor directly, update that call site to drop the argument — grep first:
Run: grep -rn "start_date=" backend/tests/ backend/app/services/campaign_scheduler_service.py
Fix any remaining call sites the grep turns up (the scheduler's _clone_campaign does not set start_date today, so it should need no change).
- Step 8: Commit
git add backend/app/routers/campaigns.py backend/app/services/campaign_service.py backend/app/services/campaign_crud_service.py backend/tests/test_campaign_approval_router.py
git commit -m "feat(campaigns): timeline endpoint, admin-only manual activation, remove lead-set dates"
Task 12: Backend — full-suite regression check
Files: none (verification only)
- Step 1: Run the complete backend test suite
Run (from backend/): pytest tests/ -v
Expected: PASS — every test green, including all new files from Tasks 1–11.
- Step 2: Run Ruff lint
Run (from backend/): ruff check app/ tests/
Expected: no errors. Fix any import-order or unused-import issues the edits introduced (e.g. if Optional or datetime end up unused anywhere after the start_date removal in Task 11).
- Step 3: Commit any lint fixes
git add -A
git commit -m "chore(campaigns): fix lint issues from approval workflow"
(Skip this commit if step 2 found nothing to fix.)
Task 13: Frontend — API client for the approval workflow
Files:
-
Modify:
frontend/src/api/campaigns.ts -
Step 1: Add new types and drop
start_datefrom create/update payloads
In frontend/src/api/campaigns.ts:
- Add
approved_by,approved_at,rejection_reasonto theCampaigninterface, right aftercompleted_at:
approved_by: string | null;
approved_at: string | null;
rejection_reason: string | null;
- Remove
start_datefromCampaignCreatePayload(it's now ignored by the backend and only ever settable viaapproveCampaign):
export interface CampaignCreatePayload {
name: string;
description?: string;
type?: string;
threat_actor_id?: string;
target_platform?: string;
tags?: string[];
scheduled_at?: string;
}
- Add the modification-request type and campaign timeline entry type at the end of the "Types" section:
export type ModificationRequestAction = "add_test" | "remove_test";
export type ModificationRequestStatus = "pending" | "approved" | "rejected";
export interface CampaignModificationRequest {
id: string;
campaign_id: string;
campaign_name: string | null;
requested_by: string | null;
action: ModificationRequestAction;
test_id: string;
test_name: string | null;
order_index: number | null;
phase: string | null;
justification: string;
status: ModificationRequestStatus;
reviewed_by: string | null;
reviewed_at: string | null;
review_notes: string | null;
created_at: string | null;
}
export interface CampaignTimelineEntry {
id: string;
action: string;
user_id: string | null;
timestamp: string;
details: Record<string, unknown>;
}
- Step 2: Add the API functions
Append to frontend/src/api/campaigns.ts:
// ── Approval workflow ──────────────────────────────────────────────
/** Submit a draft campaign into the manager's approval queue. */
export async function submitCampaignForApproval(campaignId: string): Promise<Campaign> {
const { data } = await client.post<Campaign>(`/campaigns/${campaignId}/submit`);
return data;
}
/** Manager approves a pending campaign, fixing its start date. */
export async function approveCampaign(campaignId: string, startDate: string): Promise<Campaign> {
const { data } = await client.post<Campaign>(`/campaigns/${campaignId}/approve`, {
start_date: startDate,
});
return data;
}
/** Manager rejects a pending campaign with a mandatory reason. */
export async function rejectCampaign(campaignId: string, reason: string): Promise<Campaign> {
const { data } = await client.post<Campaign>(`/campaigns/${campaignId}/reject`, { reason });
return data;
}
// ── Modification requests ───────────────────────────────────────────
/** File a request to add/remove a test on an active campaign. */
export async function createModificationRequest(
campaignId: string,
payload: {
action: ModificationRequestAction;
test_id: string;
justification: string;
order_index?: number;
phase?: string;
},
): Promise<CampaignModificationRequest> {
const { data } = await client.post<CampaignModificationRequest>(
`/campaigns/${campaignId}/modification-requests`,
payload,
);
return data;
}
/** List modification requests filed against a specific campaign. */
export async function listCampaignModificationRequests(
campaignId: string,
): Promise<CampaignModificationRequest[]> {
const { data } = await client.get<CampaignModificationRequest[]>(
`/campaigns/${campaignId}/modification-requests`,
);
return data;
}
/** Manager's global queue of modification requests awaiting review. */
export async function listPendingModificationRequests(): Promise<CampaignModificationRequest[]> {
const { data } = await client.get<CampaignModificationRequest[]>(
"/campaigns/modification-requests/pending",
);
return data;
}
/** Manager approves a modification request — the test change is applied now. */
export async function approveModificationRequest(
requestId: string,
): Promise<CampaignModificationRequest> {
const { data } = await client.post<CampaignModificationRequest>(
`/campaigns/modification-requests/${requestId}/approve`,
);
return data;
}
/** Manager rejects a modification request with mandatory review notes. */
export async function rejectModificationRequest(
requestId: string,
reviewNotes: string,
): Promise<CampaignModificationRequest> {
const { data } = await client.post<CampaignModificationRequest>(
`/campaigns/modification-requests/${requestId}/reject`,
{ review_notes: reviewNotes },
);
return data;
}
// ── Timeline ─────────────────────────────────────────────────────────
/** Get the audit-log timeline for a campaign. */
export async function getCampaignTimeline(campaignId: string): Promise<CampaignTimelineEntry[]> {
const { data } = await client.get<CampaignTimelineEntry[]>(`/campaigns/${campaignId}/timeline`);
return data;
}
- Step 3: Verify TypeScript compiles
Run (from frontend/): npm run build
Expected: build succeeds. It's expected/fine that CampaignsPage.tsx and CampaignDetailPage.tsx still reference newCampaign.start_date — those get fixed in Tasks 14–15. If npm run build fails only on those two files' start_date usage, that confirms the type change took effect correctly; proceed to Task 14.
- Step 4: Commit
git add frontend/src/api/campaigns.ts
git commit -m "feat(campaigns): API client for approval workflow"
Task 14: Frontend — CampaignsPage: drop lead-set start dates, add pending-approval filter
Files:
-
Modify:
frontend/src/pages/CampaignsPage.tsx -
Step 1: Remove the "Start date" field from the Create Campaign modal
In frontend/src/pages/CampaignsPage.tsx:
- Remove
start_date: ""from thenewCampaignstate initializer (line 58) and from thesetNewCampaignreset call increateMutation.onSuccess(line 78):
const [newCampaign, setNewCampaign] = useState({
name: "",
description: "",
type: "custom",
target_platform: "",
});
setNewCampaign({ name: "", description: "", type: "custom", target_platform: "" });
- Delete the entire "Start date"
<div className="mt-4">...</div>block (lines 232–253) from the Create Campaign modal.
- Step 2: Remove the "Start date" field from the Generate-from-Threat-Actor modal
Remove the actorStartDate state (line 52) and its setter usages:
const [actorSearch, setActorSearch] = useState("");
(delete the actorStartDate line entirely)
Update generateMutation (lines 90–99) to stop passing a start date:
const generateMutation = useMutation({
mutationFn: (actorId: string) => generateCampaignFromThreatActor(actorId),
onSuccess: (campaign) => {
queryClient.invalidateQueries({ queryKey: ["campaigns"] });
setShowActorSelector(false);
navigate(`/campaigns/${campaign.id}`);
},
});
Delete the entire "Start date" <div className="mb-4">...</div> block from the actor selector modal (the one right after the search input, roughly lines 295–316). The setActorStartDate("") call that used to live in generateMutation's onSuccess is already gone since Step 2 rewrote that whole mutation block above — no other call sites reference actorStartDate after this change.
- Step 3: Add a "Pending Approval" option to the status filter
In the status <select> (around line 166), add the option:
<option value="draft">Draft</option>
<option value="pending_approval">Pending Approval</option>
<option value="active">Active</option>
- Step 4: Add a status color/label for
pending_approval
In the statusColors map at the top of the file (line 18), add:
const statusColors: Record<string, string> = {
draft: "bg-gray-800/50 text-gray-400 border-gray-600/30",
pending_approval: "bg-amber-900/50 text-amber-400 border-amber-500/30",
active: "bg-cyan-900/50 text-cyan-400 border-cyan-500/30",
completed: "bg-green-900/50 text-green-400 border-green-500/30",
archived: "bg-gray-800/50 text-gray-500 border-gray-700/30",
};
- Step 5: Update
generateCampaignFromThreatActor's API signature usage
Confirm generateCampaignFromThreatActor in frontend/src/api/campaigns.ts still accepts an optional second argument (it does, from before this plan) — the call site from Step 2 simply omits it now, which is valid since the parameter is optional. No further change needed there.
- Step 6: Verify TypeScript compiles
Run (from frontend/): npm run build
Expected: build succeeds with no errors referencing CampaignsPage.tsx.
- Step 7: Commit
git add frontend/src/pages/CampaignsPage.tsx
git commit -m "feat(campaigns): remove lead-set start dates, add pending-approval filter"
Task 15: Frontend — CampaignDetailPage: submit / approve / reject UI
Files:
-
Modify:
frontend/src/pages/CampaignDetailPage.tsx -
Step 1: Import the new API functions and add
managerto the role checks
In frontend/src/pages/CampaignDetailPage.tsx, update the import block (lines 20–30):
import {
getCampaign,
activateCampaign,
completeCampaign,
deleteCampaign,
removeTestFromCampaign,
scheduleCampaign,
getCampaignHistory,
submitCampaignForApproval,
approveCampaign,
rejectCampaign,
type Campaign,
type CampaignHistoryEntry,
} from "../api/campaigns";
Update the role checks (lines 79–81):
const role = user?.role ?? "";
const canManage = role === "admin" || role === "red_lead" || role === "blue_lead";
const canComplete = role === "admin" || role === "red_lead";
const isManager = role === "admin" || role === "manager";
- Step 2: Add state and mutations for submit/approve/reject
Add right after the startDateWarning state (line 72):
const [approveModalOpen, setApproveModalOpen] = useState(false);
const [approveStartDate, setApproveStartDate] = useState("");
const [rejectModalOpen, setRejectModalOpen] = useState(false);
const [rejectReason, setRejectReason] = useState("");
Add right after activateMutation (after line 120):
const submitMutation = useMutation({
mutationFn: () => submitCampaignForApproval(campaignId!),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["campaign", campaignId] });
queryClient.invalidateQueries({ queryKey: ["campaign-timeline", campaignId] });
showToast("Campaign submitted for approval", "success");
},
onError: (err: Error) => showToast(err.message, "error"),
});
const approveMutation = useMutation({
mutationFn: () => approveCampaign(campaignId!, approveStartDate),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["campaign", campaignId] });
queryClient.invalidateQueries({ queryKey: ["campaign-timeline", campaignId] });
setApproveModalOpen(false);
setApproveStartDate("");
showToast("Campaign approved and activated", "success");
},
onError: (err: Error) => showToast(err.message, "error"),
});
const rejectMutation = useMutation({
mutationFn: () => rejectCampaign(campaignId!, rejectReason),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["campaign", campaignId] });
queryClient.invalidateQueries({ queryKey: ["campaign-timeline", campaignId] });
setRejectModalOpen(false);
setRejectReason("");
showToast("Campaign rejected", "success");
},
onError: (err: Error) => showToast(err.message, "error"),
});
- Step 3: Replace the "Activate" button with "Submit for Approval" (lead) and add Approve/Reject buttons (manager)
Replace the existing activate button block (lines 320–333):
{canManage && campaign.status === "draft" && (
<button
onClick={() => submitMutation.mutate()}
disabled={submitMutation.isPending}
className="flex items-center gap-1.5 rounded-lg bg-cyan-600 px-4 py-2 text-sm font-medium text-white hover:bg-cyan-500 disabled:opacity-50 transition-colors"
>
{submitMutation.isPending ? (
<Loader2 className="h-4 w-4 animate-spin" />
) : (
<Play className="h-4 w-4" />
)}
Submit for Approval
</button>
)}
{isManager && campaign.status === "pending_approval" && (
<>
<button
onClick={() => setRejectModalOpen(true)}
className="flex items-center gap-1.5 rounded-lg border border-red-500/30 bg-red-900/20 px-4 py-2 text-sm font-medium text-red-400 hover:bg-red-900/40 transition-colors"
>
Reject
</button>
<button
onClick={() => setApproveModalOpen(true)}
className="flex items-center gap-1.5 rounded-lg bg-green-600 px-4 py-2 text-sm font-medium text-white hover:bg-green-500 transition-colors"
>
<CheckCircle className="h-4 w-4" />
Approve
</button>
</>
)}
- Step 4: Show the rejection reason banner when a campaign was rejected
Add right after the campaign description block, inside the header (after the {campaign.description && (...)} block, around line 264):
{campaign.status === "draft" && campaign.rejection_reason && (
<div className="mt-2 rounded-lg border border-amber-500/30 bg-amber-900/20 px-3 py-2 text-xs text-amber-400">
Rejected by manager: {campaign.rejection_reason}
</div>
)}
- Step 5: Add the Approve and Reject modals
Add right before the "Toast notification" block (before line 723):
{/* Approve modal */}
{approveModalOpen && (
<div className="fixed inset-0 z-50 flex items-center justify-center bg-black/70 backdrop-blur-sm">
<div className="mx-4 w-full max-w-md rounded-xl border border-green-500/30 bg-gray-900 p-6 shadow-2xl">
<h3 className="mb-4 text-lg font-semibold text-white">Approve Campaign</h3>
<label className="mb-1.5 block text-sm font-medium text-gray-300">
Start date <span className="text-red-400">*</span>
</label>
<input
type="datetime-local"
value={approveStartDate}
onChange={(e) => setApproveStartDate(e.target.value)}
className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2 text-sm text-gray-200 focus:border-cyan-500 focus:outline-none [color-scheme:dark]"
/>
<div className="mt-6 flex justify-end gap-3">
<button
onClick={() => setApproveModalOpen(false)}
className="rounded-lg border border-gray-700 px-4 py-2 text-sm text-gray-400 hover:bg-gray-800 transition-colors"
>
Cancel
</button>
<button
onClick={() => approveMutation.mutate()}
disabled={!approveStartDate || approveMutation.isPending}
className="flex items-center gap-1.5 rounded-lg bg-green-600 px-4 py-2 text-sm font-medium text-white hover:bg-green-500 disabled:opacity-50 transition-colors"
>
{approveMutation.isPending && <Loader2 className="h-4 w-4 animate-spin" />}
Approve & Activate
</button>
</div>
</div>
</div>
)}
{/* Reject modal */}
{rejectModalOpen && (
<div className="fixed inset-0 z-50 flex items-center justify-center bg-black/70 backdrop-blur-sm">
<div className="mx-4 w-full max-w-md rounded-xl border border-red-500/30 bg-gray-900 p-6 shadow-2xl">
<h3 className="mb-4 text-lg font-semibold text-white">Reject Campaign</h3>
<label className="mb-1.5 block text-sm font-medium text-gray-300">
Reason <span className="text-red-400">*</span>
</label>
<textarea
value={rejectReason}
onChange={(e) => setRejectReason(e.target.value)}
rows={3}
placeholder="Explain what needs to change before this can be approved…"
className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2 text-sm text-gray-200 placeholder-gray-500 focus:border-cyan-500 focus:outline-none"
/>
<div className="mt-6 flex justify-end gap-3">
<button
onClick={() => setRejectModalOpen(false)}
className="rounded-lg border border-gray-700 px-4 py-2 text-sm text-gray-400 hover:bg-gray-800 transition-colors"
>
Cancel
</button>
<button
onClick={() => rejectMutation.mutate()}
disabled={!rejectReason.trim() || rejectMutation.isPending}
className="flex items-center gap-1.5 rounded-lg bg-red-600 px-4 py-2 text-sm font-medium text-white hover:bg-red-500 disabled:opacity-50 transition-colors"
>
{rejectMutation.isPending && <Loader2 className="h-4 w-4 animate-spin" />}
Reject
</button>
</div>
</div>
</div>
)}
- Step 6: Verify TypeScript compiles
Run (from frontend/): npm run build
Expected: build succeeds.
- Step 7: Manual verification in the browser
Start the dev server (preview_start with the frontend config, or npm run dev inside frontend/). As a red_lead user, create a draft campaign, add a test, click "Submit for Approval," confirm the status badge changes to pending_approval and the Activate button is gone. Log in as a manager (create one via Users page if needed) and confirm the Approve/Reject buttons appear, that Approve requires a date, and that Reject requires a reason and returns the campaign to draft with the banner visible.
- Step 8: Commit
git add frontend/src/pages/CampaignDetailPage.tsx
git commit -m "feat(campaigns): submit/approve/reject UI on campaign detail page"
Task 16: Frontend — modification request UI
Files:
-
Create:
frontend/src/components/RequestCampaignModificationModal.tsx -
Modify:
frontend/src/pages/CampaignDetailPage.tsx -
Step 1: Create the request modal component
Create frontend/src/components/RequestCampaignModificationModal.tsx:
import { useState } from "react";
import { useMutation, useQueryClient } from "@tanstack/react-query";
import { X, Loader2 } from "lucide-react";
import { createModificationRequest, type CampaignTest } from "../api/campaigns";
interface Props {
campaignId: string;
tests: CampaignTest[];
open: boolean;
onClose: () => void;
onSuccess: () => void;
}
export default function RequestCampaignModificationModal({
campaignId,
tests,
open,
onClose,
onSuccess,
}: Props) {
const queryClient = useQueryClient();
const [action, setAction] = useState<"add_test" | "remove_test">("remove_test");
const [testId, setTestId] = useState("");
const [justification, setJustification] = useState("");
const mutation = useMutation({
mutationFn: () =>
createModificationRequest(campaignId, {
action,
test_id: testId,
justification: justification.trim(),
}),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["campaign-modification-requests", campaignId] });
setTestId("");
setJustification("");
onSuccess();
onClose();
},
});
if (!open) return null;
const canSubmit = testId && justification.trim().length > 0 && !mutation.isPending;
return (
<div className="fixed inset-0 z-50 flex items-center justify-center bg-black/60 backdrop-blur-sm">
<div className="w-full max-w-lg rounded-xl border border-gray-800 bg-gray-900 p-6">
<div className="mb-4 flex items-center justify-between">
<h2 className="text-lg font-semibold text-white">Request Campaign Modification</h2>
<button onClick={onClose} className="rounded-lg p-1.5 text-gray-400 hover:bg-gray-800 hover:text-white">
<X className="h-5 w-5" />
</button>
</div>
<div className="space-y-4">
<div>
<label className="mb-1.5 block text-sm font-medium text-gray-300">Action</label>
<select
value={action}
onChange={(e) => setAction(e.target.value as "add_test" | "remove_test")}
className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2 text-sm text-gray-300 focus:border-cyan-500 focus:outline-none"
>
<option value="remove_test">Remove a test from this campaign</option>
<option value="add_test">Add an existing test to this campaign</option>
</select>
</div>
{action === "remove_test" ? (
<div>
<label className="mb-1.5 block text-sm font-medium text-gray-300">Test to remove</label>
<select
value={testId}
onChange={(e) => setTestId(e.target.value)}
className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2 text-sm text-gray-300 focus:border-cyan-500 focus:outline-none"
>
<option value="">Select a test…</option>
{tests.map((t) => (
<option key={t.test_id} value={t.test_id}>
{t.test_name || t.test_id}
</option>
))}
</select>
</div>
) : (
<div>
<label className="mb-1.5 block text-sm font-medium text-gray-300">Test ID to add</label>
<input
value={testId}
onChange={(e) => setTestId(e.target.value)}
placeholder="Paste the existing test's ID"
className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2 text-sm text-gray-200 placeholder-gray-500 focus:border-cyan-500 focus:outline-none"
/>
</div>
)}
<div>
<label className="mb-1.5 block text-sm font-medium text-gray-300">
Justification <span className="text-red-400">*</span>
</label>
<textarea
value={justification}
onChange={(e) => setJustification(e.target.value)}
rows={3}
placeholder="Why does this campaign need to change?"
className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2 text-sm text-gray-200 placeholder-gray-500 focus:border-cyan-500 focus:outline-none"
/>
</div>
{mutation.isError && (
<div className="rounded-lg border border-red-500/30 bg-red-900/20 p-3 text-sm text-red-400">
{(mutation.error as Error)?.message || "Failed to submit request"}
</div>
)}
</div>
<div className="mt-6 flex justify-end gap-3">
<button
onClick={onClose}
className="rounded-lg border border-gray-700 px-4 py-2 text-sm text-gray-400 hover:bg-gray-800 transition-colors"
>
Cancel
</button>
<button
onClick={() => mutation.mutate()}
disabled={!canSubmit}
className="flex items-center gap-1.5 rounded-lg bg-cyan-600 px-4 py-2 text-sm font-medium text-white hover:bg-cyan-500 disabled:opacity-50 transition-colors"
>
{mutation.isPending && <Loader2 className="h-4 w-4 animate-spin" />}
Submit Request
</button>
</div>
</div>
</div>
);
}
- Step 2: Wire the modal and the pending-requests panel into CampaignDetailPage
In frontend/src/pages/CampaignDetailPage.tsx, add imports:
import {
listCampaignModificationRequests,
approveModificationRequest,
rejectModificationRequest,
} from "../api/campaigns";
import RequestCampaignModificationModal from "../components/RequestCampaignModificationModal";
Add state (near showAddTestModal):
const [showModRequestModal, setShowModRequestModal] = useState(false);
Add the query and mutations (near historyData's query):
const { data: modRequests = [] } = useQuery({
queryKey: ["campaign-modification-requests", campaignId],
queryFn: () => listCampaignModificationRequests(campaignId!),
enabled: !!campaignId,
});
const approveModRequestMutation = useMutation({
mutationFn: (requestId: string) => approveModificationRequest(requestId),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["campaign", campaignId] });
queryClient.invalidateQueries({ queryKey: ["campaign-modification-requests", campaignId] });
showToast("Modification approved and applied", "success");
},
onError: (err: Error) => showToast(err.message, "error"),
});
const rejectModRequestMutation = useMutation({
mutationFn: ({ requestId, notes }: { requestId: string; notes: string }) =>
rejectModificationRequest(requestId, notes),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["campaign-modification-requests", campaignId] });
showToast("Modification rejected", "success");
},
onError: (err: Error) => showToast(err.message, "error"),
});
Add a "Request Modification" button next to the Tests table header, right after the existing {canManage && campaign.status === "draft" && (...)} Add Test button (around line 585):
{canManage && campaign.status === "active" && (
<button
onClick={() => setShowModRequestModal(true)}
className="flex items-center gap-1.5 rounded-lg border border-amber-500/30 bg-amber-500/10 px-3 py-1.5 text-sm font-medium text-amber-400 hover:bg-amber-500/20 transition-colors"
>
Request Modification
</button>
)}
Add a pending-modification-requests panel right after the Tests Table section closes (after line 677, before the Jira panel at line 680):
{/* Modification Requests */}
{modRequests.length > 0 && (
<div className="rounded-xl border border-gray-800 bg-gray-900 p-6">
<h2 className="mb-4 text-lg font-semibold text-white">Modification Requests</h2>
<div className="space-y-3">
{modRequests.map((r) => (
<div key={r.id} className="rounded-lg border border-gray-800 p-4">
<div className="flex items-start justify-between gap-4">
<div>
<p className="text-sm text-gray-200">
<span className="font-medium capitalize">{r.action.replace("_", " ")}</span>
{" — "}
{r.test_name || r.test_id}
</p>
<p className="mt-1 text-xs text-gray-400">{r.justification}</p>
{r.review_notes && (
<p className="mt-1 text-xs text-amber-400">Manager notes: {r.review_notes}</p>
)}
</div>
<span
className={`shrink-0 rounded-full border px-2 py-0.5 text-[10px] font-medium ${
r.status === "pending"
? "border-amber-500/30 bg-amber-900/30 text-amber-400"
: r.status === "approved"
? "border-green-500/30 bg-green-900/30 text-green-400"
: "border-red-500/30 bg-red-900/30 text-red-400"
}`}
>
{r.status}
</span>
</div>
{isManager && r.status === "pending" && (
<div className="mt-3 flex gap-2">
<button
onClick={() => approveModRequestMutation.mutate(r.id)}
className="rounded-lg bg-green-600 px-3 py-1.5 text-xs font-medium text-white hover:bg-green-500 transition-colors"
>
Approve
</button>
<button
onClick={() => {
const notes = window.prompt("Reason for rejecting this request:");
if (notes && notes.trim()) {
rejectModRequestMutation.mutate({ requestId: r.id, notes: notes.trim() });
}
}}
className="rounded-lg border border-red-500/30 bg-red-900/20 px-3 py-1.5 text-xs font-medium text-red-400 hover:bg-red-900/40 transition-colors"
>
Reject
</button>
</div>
)}
</div>
))}
</div>
</div>
)}
{/* Request Modification Modal */}
<RequestCampaignModificationModal
campaignId={campaignId!}
tests={campaign.tests}
open={showModRequestModal}
onClose={() => setShowModRequestModal(false)}
onSuccess={() => showToast("Modification request submitted", "success")}
/>
- Step 3: Verify TypeScript compiles
Run (from frontend/): npm run build
Expected: build succeeds.
- Step 4: Manual verification in the browser
As a red_lead on an active campaign, click "Request Modification," pick "Remove a test," fill justification, submit. Confirm the test is still in the Tests table (unchanged) and a new "pending" entry appears under Modification Requests. Log in as manager, click Approve on that entry, confirm the test disappears from the Tests table and the request shows "approved."
- Step 5: Commit
git add frontend/src/components/RequestCampaignModificationModal.tsx frontend/src/pages/CampaignDetailPage.tsx
git commit -m "feat(campaigns): modification request UI on campaign detail page"
Task 17: Frontend — timeline tab and role list updates
Files:
-
Create:
frontend/src/components/CampaignAuditTimeline.tsx -
Modify:
frontend/src/pages/CampaignDetailPage.tsx -
Modify:
frontend/src/pages/UsersPage.tsx -
Modify:
frontend/src/pages/SettingsPage.tsx -
Step 1: Create the audit timeline component
Create frontend/src/components/CampaignAuditTimeline.tsx:
import { Clock, Loader2 } from "lucide-react";
import type { CampaignTimelineEntry } from "../api/campaigns";
interface Props {
entries: CampaignTimelineEntry[];
isLoading: boolean;
}
const ACTION_LABELS: Record<string, string> = {
create_campaign: "Campaign created",
update_campaign: "Campaign updated",
submit_campaign_for_approval: "Submitted for approval",
approve_campaign: "Approved by manager",
reject_campaign: "Rejected by manager",
activate_campaign: "Activated",
complete_campaign: "Marked completed",
request_campaign_modification: "Modification requested",
approve_campaign_modification: "Modification approved",
reject_campaign_modification: "Modification rejected",
};
export default function CampaignAuditTimeline({ entries, isLoading }: Props) {
if (isLoading) {
return (
<div className="flex items-center justify-center py-12">
<Loader2 className="h-6 w-6 animate-spin text-gray-500" />
</div>
);
}
if (entries.length === 0) {
return (
<div className="py-8 text-center">
<Clock className="mx-auto h-8 w-8 text-gray-600" />
<p className="mt-2 text-sm text-gray-400">No timeline events yet</p>
</div>
);
}
const formatDate = (d: string) =>
new Date(d).toLocaleString("en-US", {
month: "short",
day: "numeric",
hour: "2-digit",
minute: "2-digit",
second: "2-digit",
});
return (
<div className="relative space-y-0">
<div className="absolute left-4 top-2 bottom-2 w-px bg-gray-700" />
{entries.map((entry, idx) => (
<div key={entry.id || idx} className="relative flex gap-4 py-3">
<div className="z-10 flex h-8 w-8 shrink-0 items-center justify-center rounded-full border border-gray-700 bg-gray-800">
<Clock className="h-3.5 w-3.5 text-gray-400" />
</div>
<div className="flex-1 pt-0.5">
<p className="text-sm font-medium text-gray-200">
{ACTION_LABELS[entry.action] || entry.action}
</p>
<p className="text-xs text-gray-500">{formatDate(entry.timestamp)}</p>
</div>
</div>
))}
</div>
);
}
- Step 2: Wire the timeline into CampaignDetailPage
Add imports:
import { getCampaignTimeline } from "../api/campaigns";
import CampaignAuditTimeline from "../components/CampaignAuditTimeline";
Add the query (near historyData):
const { data: timeline = [], isLoading: isTimelineLoading } = useQuery({
queryKey: ["campaign-timeline", campaignId],
queryFn: () => getCampaignTimeline(campaignId!),
enabled: !!campaignId,
});
Add a Timeline section right after the Modification Requests panel added in Task 16 (or right before the Jira panel if Task 16 wasn't applicable):
{/* Timeline */}
<div className="rounded-xl border border-gray-800 bg-gray-900 p-6">
<h2 className="mb-4 text-lg font-semibold text-white">Timeline</h2>
<CampaignAuditTimeline entries={timeline} isLoading={isTimelineLoading} />
</div>
- Step 3: Add
managerto the Users page role list
In frontend/src/pages/UsersPage.tsx, update ROLES (line 17):
const ROLES = [
{ value: "viewer", label: "Viewer" },
{ value: "red_tech", label: "Red Tech" },
{ value: "blue_tech", label: "Blue Tech" },
{ value: "red_lead", label: "Red Lead" },
{ value: "blue_lead", label: "Blue Lead" },
{ value: "manager", label: "Manager" },
{ value: "admin", label: "Admin" },
];
Update roleBadgeColors (line 26):
const roleBadgeColors: Record<string, string> = {
admin: "bg-purple-900/50 text-purple-400 border-purple-500/30",
red_tech: "bg-red-900/50 text-red-400 border-red-500/30",
blue_tech: "bg-blue-900/50 text-blue-400 border-blue-500/30",
red_lead: "bg-orange-900/50 text-orange-400 border-orange-500/30",
blue_lead: "bg-cyan-900/50 text-cyan-400 border-cyan-500/30",
manager: "bg-amber-900/50 text-amber-400 border-amber-500/30",
viewer: "bg-gray-800/50 text-gray-400 border-gray-600/30",
};
- Step 4: Add
managerto the SSO default-role dropdown
In frontend/src/pages/SettingsPage.tsx, find the role options array around line 1217 and add manager between blue_lead and red_tech (or any consistent position):
{ value: "admin", label: "Aegis Admin", desc: "Full platform access including system settings" },
{ value: "red_lead", label: "Aegis Red Lead", desc: "Red team lead — manage tests, campaigns and templates" },
{ value: "blue_lead", label: "Aegis Blue Lead", desc: "Blue team lead — validate tests and manage coverage" },
{ value: "manager", label: "Aegis Manager", desc: "Approves campaigns and campaign changes" },
{ value: "red_tech", label: "Aegis Red Tech", desc: "Red team technician — execute tests" },
{ value: "blue_tech", label: "Aegis Blue Tech", desc: "Blue team technician — review detections" },
{ value: "viewer", label: "Aegis Viewer", desc: "Read-only access to dashboards and reports" },
- Step 5: Verify TypeScript compiles
Run (from frontend/): npm run build
Expected: build succeeds with zero errors.
- Step 6: Manual verification in the browser
As admin, go to Users, create a user with role "Manager," confirm the badge renders correctly. Open a campaign with some audit history (submit/approve/reject a test campaign) and confirm the Timeline section lists the actions in order with readable labels.
- Step 7: Commit
git add frontend/src/components/CampaignAuditTimeline.tsx frontend/src/pages/CampaignDetailPage.tsx frontend/src/pages/UsersPage.tsx frontend/src/pages/SettingsPage.tsx
git commit -m "feat(campaigns): audit timeline UI and manager role in role pickers"
Task 18: Final end-to-end verification
Files: none (verification only)
- Step 1: Run the full backend suite one more time
Run (from backend/): pytest tests/ -v
Expected: 100% pass.
- Step 2: Run the frontend build one more time
Run (from frontend/): npm run build
Expected: build succeeds.
- Step 3: Full manual walkthrough in the browser
Using the preview tools (or npm run dev):
- As
red_lead: create a campaign, add a test, submit for approval. Confirm you cannot see an Activate button and cannot edit start date anywhere. - As
manager: see the campaign show up when filtering by "Pending Approval" on the Campaigns list, open it, approve it with a start date. Confirm it becomesactivewith that date shown. - As
red_lead: on the now-active campaign, request removing the test with a justification. Confirm the test is still present. - As
manager: approve the modification request. Confirm the test disappears from the campaign and the request shows "approved." - On the campaign detail page, check the Timeline section shows: created → submitted for approval → approved → modification requested → modification approved, in that order.
- As
red_tech: confirm you get no Submit/Approve/Request-Modification buttons anywhere (view-only).
- Step 4: Report completion
No commit for this task — it's verification only. If any step in the walkthrough fails, return to the relevant task above and fix before considering the feature done.