Spaces:
Sleeping
Sleeping
File size: 3,652 Bytes
e9d2669 f23b079 cc6cee9 f23b079 cc6cee9 f23b079 910cb0b 726069c e9d2669 792794b 5c5bfec d41dc87 3d53300 c1dbaf4 792794b e9d2669 910cb0b 68bc590 f23b079 792794b 726069c e9d2669 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | from __future__ import annotations
from dataclasses import asdict, dataclass, field
from typing import Any
@dataclass
class AgentRuntimeConfig:
agent_key: str
agent_name: str
execution_mode: str
timeout_seconds: int = 60
provider_strategy: str = ""
model_family: str = ""
model_override: str = ""
model_id: str = ""
directives: str = ""
@dataclass
class RunControlConfig:
max_rounds: int
max_feedback_messages: int
max_feedback_per_agent_pair: int
@dataclass
class WorkingMemory:
shared: dict[str, Any] = field(default_factory=dict)
agent_private: dict[str, dict[str, Any]] = field(default_factory=dict)
timeline: list[str] = field(default_factory=list)
def read_shared(self, key: str, default: Any = None) -> Any:
return self.shared.get(key, default)
def write_shared(self, key: str, value: Any, *, author: str | None = None) -> None:
self.shared[key] = value
if author:
self.timeline.append(f"{author} wrote shared memory key '{key}'.")
def read_agent(self, agent_key: str, key: str, default: Any = None) -> Any:
return self.agent_private.get(agent_key, {}).get(key, default)
def write_agent(self, agent_key: str, key: str, value: Any) -> None:
self.agent_private.setdefault(agent_key, {})[key] = value
self.timeline.append(f"{agent_key} updated private memory key '{key}'.")
def add_note(self, note: str) -> None:
if note.strip():
self.timeline.append(note.strip())
@dataclass
class RunSession:
title: str
source_requirements: str
working_memory: WorkingMemory = field(default_factory=WorkingMemory)
@dataclass
class RequirementItem:
requirement_id: str
original_text: str
normalized_text: str
priority: str
acceptance_criteria: list[str]
assumptions: list[str] = field(default_factory=list)
@dataclass
class TestCaseDesign:
test_case_id: str
requirement_id: str
title: str
test_type: str
preconditions: list[str]
steps: list[str]
expected_results: list[str]
oracle: str
risks: list[str] = field(default_factory=list)
@dataclass
class GeneratedArtifact:
artifact_id: str
requirement_id: str
design_id: str
target: str
test_name: str
test_data: dict[str, Any]
selectors: list[str]
pseudocode: list[str]
@dataclass
class ReviewReport:
approved: bool
coverage_ratio: float
findings: list[str]
improvement_actions: list[str]
@dataclass
class StageTrace:
iteration: int
stage_index: int
agent_name: str
input_summary: list[str]
output_summary: list[str]
status: str
reasoning_trace: list[str] = field(default_factory=list)
reasoning_source: str = "structured_trace"
agent_explanation: str = ""
decision_explanation: str = ""
duration_ms: int = 0
prompt_tokens: int | None = None
completion_tokens: int | None = None
total_tokens: int | None = None
@dataclass
class PipelineResult:
title: str
source_requirements: str
requirements: list[RequirementItem]
test_designs: list[TestCaseDesign]
generated_artifacts: list[GeneratedArtifact]
review: ReviewReport
iterations: int
run_controls: RunControlConfig
total_feedback_messages: int = 0
pair_feedback_counts: dict[str, int] = field(default_factory=dict)
agent_configs: list[AgentRuntimeConfig] = field(default_factory=list)
stage_traces: list[StageTrace] = field(default_factory=list)
run_session: RunSession | None = None
def to_dict(self) -> dict[str, Any]:
return asdict(self)
|