gongzhiyongandOmX
c22dc7f573
Make the user's six swarm characteristics first-class acceptance gates by adding S10/A07 tests, a standard document, and synchronized reports. Constraint: The user asked to set acceptance indicators and test standard details around decentralization, self-organization, emergence, robustness, scalability, and implicit collaboration. Rejected: Treating the six traits as prose-only documentation | they now run as deterministic tests and scenario matrix gates. Confidence: high Scope-risk: moderate Directive: Future swarm-readiness claims must report F01-F06 explicitly and distinguish local Agent-layer proof from production no-coordinator runtime. Tested: py_compile; unittest discover ran 41 tests; run_swarm_characteristics_acceptance PASS; run_academic_standard_evaluation A01-A07 PASS; run_standard_scenario_acceptance S01-S10 PASS with S07 run_id 9c7ccc6087c1435694a52efb12c32301; docs/README secret-pattern scan clean; git diff --cached --check clean. Not-tested: Production no-coordinator distributed runtime and Kubernetes-scale worker telemetry remain outside this minimal local acceptance gate. Co-authored-by: OmX <omx@oh-my-codex.dev>
203 lines
10 KiB
Python
203 lines
10 KiB
Python
import unittest
|
|
|
|
from swarm_minimal.core import Agent, InMemorySwarmStore, SwarmCoordinator, Task, TaskStatus
|
|
|
|
|
|
class SwarmCharacteristicsAcceptanceTest(unittest.TestCase):
|
|
def test_decentralization_has_no_single_agent_control_node(self) -> None:
|
|
store = InMemorySwarmStore()
|
|
run_id = "feature-decentralization"
|
|
store.shared_state[f"run:{run_id}:goal"] = "agent-level decentralized claim"
|
|
store.shared_state[f"run:{run_id}:status"] = "running"
|
|
for index in range(8):
|
|
store.add_task(Task(kind="autonomous", input=f"local decision {index}"))
|
|
|
|
agents = [
|
|
Agent(
|
|
id=f"autonomous-agent-{index}",
|
|
capability="autonomous",
|
|
run=lambda task, shared_state, index=index: (
|
|
shared_state.setdefault(f"decision:{index}:{task.id}", f"agent={index}; task={task.input}")
|
|
or f"agent={index}; task={task.input}",
|
|
0.7 + index / 100,
|
|
),
|
|
)
|
|
for index in range(4)
|
|
]
|
|
|
|
report = SwarmCoordinator(store=store, agents=agents).run_autonomous_until_converged(run_id)
|
|
participating_agents = {event.agent_id for event in report.claim_events}
|
|
decision_keys = [key for key in store.shared_state if key.startswith("decision:")]
|
|
control_keys = [key for key in store.shared_state if "leader" in key or "controller" in key]
|
|
|
|
self.assertTrue(report.converged)
|
|
self.assertEqual(report.duplicate_claims, ())
|
|
self.assertEqual(len(participating_agents), 4)
|
|
self.assertEqual(len(decision_keys), 8)
|
|
self.assertEqual(control_keys, [])
|
|
|
|
def test_self_organization_forms_order_from_local_interactions(self) -> None:
|
|
store = InMemorySwarmStore()
|
|
run_id = "feature-self-organization"
|
|
store.shared_state[f"run:{run_id}:goal"] = "local signals form ordered cluster"
|
|
store.shared_state[f"run:{run_id}:status"] = "running"
|
|
for payload in ["api:0.31", "docs:0.22", "api:0.29", "tests:0.18", "api:0.27"]:
|
|
store.add_task(Task(kind="organize", input=payload))
|
|
|
|
def organize(task: Task, shared_state: dict[str, str]) -> tuple[str, float]:
|
|
cluster, value_text = task.input.split(":")
|
|
value = float(value_text)
|
|
count_key = f"cluster:{cluster}:count"
|
|
score_key = f"cluster:{cluster}:score"
|
|
count = int(shared_state.get(count_key, "0")) + 1
|
|
score = float(shared_state.get(score_key, "0")) + value
|
|
shared_state[count_key] = str(count)
|
|
shared_state[score_key] = f"{score:.2f}"
|
|
candidates = {
|
|
key.removeprefix("cluster:").removesuffix(":score"): float(item)
|
|
for key, item in shared_state.items()
|
|
if key.startswith("cluster:") and key.endswith(":score")
|
|
}
|
|
dominant = max(candidates.items(), key=lambda item: item[1])[0]
|
|
shared_state[f"run:{run_id}:dominant_cluster"] = dominant
|
|
return f"cluster={dominant}; local={cluster}; count={count}; score={score:.2f}", score
|
|
|
|
agents = [Agent(id=f"organizer-{index}", capability="organize", run=organize) for index in range(3)]
|
|
|
|
result = SwarmCoordinator(store=store, agents=agents).run_until_converged(run_id)
|
|
|
|
self.assertEqual(result.completed_tasks, 5)
|
|
self.assertEqual(store.shared_state[f"run:{run_id}:dominant_cluster"], "api")
|
|
self.assertEqual(store.shared_state["cluster:api:count"], "3")
|
|
self.assertGreater(float(store.shared_state["cluster:api:score"]), float(store.shared_state["cluster:docs:score"]))
|
|
self.assertIn("cluster=api", result.accepted_output)
|
|
|
|
def test_emergence_global_result_exceeds_single_local_signal(self) -> None:
|
|
store = InMemorySwarmStore()
|
|
run_id = "feature-emergence"
|
|
store.shared_state[f"run:{run_id}:goal"] = "global behavior from local evidence"
|
|
store.shared_state[f"run:{run_id}:status"] = "running"
|
|
for payload in ["alpha:0.31", "beta:0.33", "beta:0.34", "gamma:0.45"]:
|
|
store.add_task(Task(kind="evidence", input=payload))
|
|
|
|
local_scores: list[float] = []
|
|
|
|
def contribute(task: Task, shared_state: dict[str, str]) -> tuple[str, float]:
|
|
candidate, value_text = task.input.split(":")
|
|
value = float(value_text)
|
|
local_scores.append(value)
|
|
key = f"candidate:{candidate}:score"
|
|
total = float(shared_state.get(key, "0")) + value
|
|
shared_state[key] = f"{total:.2f}"
|
|
return f"candidate={candidate}; local={value:.2f}; group_total={total:.2f}", total
|
|
|
|
agents = [Agent(id=f"evidence-agent-{index}", capability="evidence", run=contribute) for index in range(3)]
|
|
|
|
result = SwarmCoordinator(store=store, agents=agents).run_until_converged(run_id)
|
|
|
|
self.assertIn("candidate=beta", result.accepted_output)
|
|
self.assertEqual(store.shared_state["candidate:beta:score"], "0.67")
|
|
self.assertGreater(result.accepted_score, max(local_scores))
|
|
|
|
def test_robustness_single_agent_failure_does_not_stop_convergence(self) -> None:
|
|
store = InMemorySwarmStore()
|
|
run_id = "feature-robustness"
|
|
store.shared_state[f"run:{run_id}:goal"] = "one failed route should not stop swarm"
|
|
store.shared_state[f"run:{run_id}:status"] = "running"
|
|
for payload in ["fragile-route", "robust-route-a", "robust-route-b"]:
|
|
store.add_task(Task(kind="route", input=payload))
|
|
|
|
agents = [
|
|
Agent(
|
|
id="crashing-agent",
|
|
capability="route",
|
|
run=lambda task, _: (_ for _ in ()).throw(RuntimeError("forced individual failure")),
|
|
),
|
|
Agent(id="backup-agent-a", capability="route", run=lambda task, _: ("healthy result", 0.91)),
|
|
Agent(id="backup-agent-b", capability="route", run=lambda task, _: ("alternative result", 0.86)),
|
|
]
|
|
|
|
result = SwarmCoordinator(store=store, agents=agents).run_until_converged(run_id)
|
|
failed = [task for task in store.tasks.values() if task.status == TaskStatus.FAILED]
|
|
done = [task for task in store.tasks.values() if task.status == TaskStatus.DONE]
|
|
|
|
self.assertEqual(len(failed), 1)
|
|
self.assertEqual(len(done), 2)
|
|
self.assertEqual(result.completed_tasks, 2)
|
|
self.assertEqual(store.shared_state[f"run:{run_id}:status"], "converged")
|
|
self.assertTrue(any(observation.score_delta < 0 for observation in result.observations))
|
|
|
|
def test_scalability_three_five_seven_agents_keep_same_architecture(self) -> None:
|
|
for agent_count in (3, 5, 7):
|
|
with self.subTest(agent_count=agent_count):
|
|
store = InMemorySwarmStore()
|
|
run_id = f"feature-scalability-{agent_count}"
|
|
store.shared_state[f"run:{run_id}:goal"] = "same architecture scaled agent count"
|
|
store.shared_state[f"run:{run_id}:status"] = "running"
|
|
for index in range(agent_count * 2):
|
|
store.add_task(Task(kind="scale", input=f"scaled task {index}"))
|
|
|
|
agents = [
|
|
Agent(
|
|
id=f"scale-agent-{index}",
|
|
capability="scale",
|
|
run=lambda task, shared_state, index=index: (
|
|
f"agent={index}; task={task.input}; architecture=shared_task_pool",
|
|
0.72 + index / 100,
|
|
),
|
|
)
|
|
for index in range(agent_count)
|
|
]
|
|
|
|
report = SwarmCoordinator(store=store, agents=agents).run_autonomous_until_converged(run_id)
|
|
|
|
self.assertTrue(report.converged)
|
|
self.assertEqual(report.completed_tasks, agent_count * 2)
|
|
self.assertEqual(report.failed_tasks, 0)
|
|
self.assertEqual(report.duplicate_claims, ())
|
|
self.assertEqual(len({event.agent_id for event in report.claim_events}), agent_count)
|
|
|
|
def test_implicit_collaboration_uses_environment_not_direct_messages(self) -> None:
|
|
store = InMemorySwarmStore()
|
|
run_id = "feature-implicit-collaboration"
|
|
store.shared_state[f"run:{run_id}:goal"] = "stigmergy through environment"
|
|
store.shared_state[f"run:{run_id}:status"] = "running"
|
|
low = Task(kind="stigmergy", input="low-signal")
|
|
high = Task(kind="stigmergy", input="high-signal")
|
|
medium = Task(kind="stigmergy", input="medium-signal")
|
|
for task in [low, high, medium]:
|
|
store.add_task(task)
|
|
store.pheromones[low.id] = 0.1
|
|
store.pheromones[high.id] = 0.9
|
|
store.pheromones[medium.id] = 0.4
|
|
claim_order: list[str] = []
|
|
|
|
def follow_environment(task: Task, shared_state: dict[str, str]) -> tuple[str, float]:
|
|
claim_order.append(task.input)
|
|
prior_trail = shared_state.get("environment:trail", "")
|
|
shared_state["environment:trail"] = (prior_trail + ">" + task.input).strip(">")
|
|
score_by_input = {
|
|
"high-signal": 0.82,
|
|
"medium-signal": 0.62,
|
|
"low-signal": 0.31,
|
|
}
|
|
return f"followed_environment={task.input}; prior_trail={prior_trail}", score_by_input[task.input]
|
|
|
|
agents = [
|
|
Agent(id="stigmergy-agent-a", capability="stigmergy", run=follow_environment),
|
|
Agent(id="stigmergy-agent-b", capability="stigmergy", run=follow_environment),
|
|
]
|
|
|
|
result = SwarmCoordinator(store=store, agents=agents).run_until_converged(run_id)
|
|
direct_message_keys = [key for key in store.shared_state if key.startswith("message:")]
|
|
|
|
self.assertEqual(claim_order[0], "high-signal")
|
|
self.assertIn("high-signal", result.accepted_output)
|
|
self.assertEqual(direct_message_keys, [])
|
|
self.assertEqual(store.shared_state["environment:trail"].split(">")[0], "high-signal")
|
|
self.assertGreater(store.pheromones[high.id], store.pheromones[medium.id])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|