Files
fengqun/tests/test_markov_process_properties.py
gongzhiyongandOmX 10a980b0bf Establish agent swarm quality evidence
Define Agent and swarm-specific acceptance evidence, move the reports under docs, and make the homepage point to the current standard, live run, model I/O, and handoff evidence.

Constraint: Agent quality standards are configured from industry AI and agent risk references because there is no single accepted swarm-Agent certification standard.

Rejected: Treating py_compile or unittest as the primary quality standard | they are evidence collection tools, not the Agent quality standard itself.

Confidence: high

Scope-risk: moderate

Directive: Keep future standard reports under docs/ and keep secrets in ignored local .env files only.

Tested: git diff --cached --check; python -B -m py_compile swarm_minimal/*.py examples/*.py tests/*.py; python -B -m unittest discover -s tests; python -u -B examples/run_academic_standard_evaluation.py

Not-tested: Did not rerun the full live Azure/NewAPI S07 scenario after moving docs; previous live run 3e8e58ae4e084bc8b90cf5c46f8992f3 passed before the docs relocation.

Co-authored-by: OmX <omx@oh-my-codex.dev>
2026-05-16 14:36:47 +08:00

73 lines
3.1 KiB
Python

import unittest
from swarm_minimal.academic_evaluation import assess_markov_process_fit
from swarm_minimal.core import Agent, InMemorySwarmStore, Observation, Task, TaskStatus
def build_equivalent_claim_store(history_label: str) -> InMemorySwarmStore:
store = InMemorySwarmStore()
low = Task(kind="probe", input="low", id="task-low")
high = Task(kind="probe", input="high", id="task-high")
store.add_task(low)
store.add_task(high)
store.pheromones[low.id] = 0.2
store.pheromones[high.id] = 0.8
store.shared_state["irrelevant_history_label"] = history_label
store.observations.append(
Observation(
task_id=f"past-{history_label}",
agent_id="past-agent",
signal="past:done",
score_delta=0.1,
)
)
return store
class MarkovProcessPropertyTest(unittest.TestCase):
def test_claim_transition_uses_current_task_pheromone_state_not_past_path(self) -> None:
agent = Agent(id="probe-agent", capability="probe", run=lambda task, _: ("ok", 0.1))
first = build_equivalent_claim_store("path-a")
second = build_equivalent_claim_store("path-b")
first_claim = first.claim_next(agent)
second_claim = second.claim_next(agent)
self.assertIsNotNone(first_claim)
self.assertIsNotNone(second_claim)
self.assertEqual(first_claim.id, "task-high")
self.assertEqual(second_claim.id, "task-high")
self.assertEqual(first.shared_state["task:task-high:claimed_by"], "probe-agent")
self.assertEqual(second.shared_state["task:task-high:claimed_by"], "probe-agent")
def test_score_update_depends_on_current_task_agent_and_output(self) -> None:
agent = Agent(id="probe-agent", capability="probe", run=lambda task, _: ("ok", 0.4))
first = build_equivalent_claim_store("path-a")
second = build_equivalent_claim_store("path-b")
first_task = first.claim_next(agent)
second_task = second.claim_next(agent)
assert first_task is not None
assert second_task is not None
first.complete_task(first_task, agent, "same output", 0.4)
second.complete_task(second_task, agent, "same output", 0.4)
self.assertEqual(first_task.status, TaskStatus.DONE)
self.assertEqual(second_task.status, TaskStatus.DONE)
self.assertEqual(first_task.output, second_task.output)
self.assertEqual(first_task.score, second_task.score)
self.assertEqual(first.pheromones[first_task.id], second.pheromones[second_task.id])
def test_project_is_markov_style_state_machine_not_formal_mdp(self) -> None:
assessment = assess_markov_process_fit()
self.assertTrue(assessment.markov_style_state_machine)
self.assertFalse(assessment.formal_markov_process)
self.assertFalse(assessment.formal_markov_decision_process)
self.assertIn("pheromone score table", assessment.sufficient_state)
self.assertTrue(any("transition probability" in item for item in assessment.limiting_factors))
if __name__ == "__main__":
unittest.main()