From 6c1358e8916a545f90cdd5d802d70de1399611ec Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Mon, 5 Oct 2026 16:12:57 +0800 Subject: [PATCH 1/2] feat: bind employee memory to pinned runs and capture coordinated learning --- docs/proposals/music-studio-template.md | 18 +- docs/roadmap.md | 6 +- .../application/memory_runtime_services.py | 359 ++++++++++----- .../organizational_memory_services.py | 3 +- tests/test_employee_memory_runtime_binding.py | 434 ++++++++++++++++++ 5 files changed, 709 insertions(+), 111 deletions(-) create mode 100644 tests/test_employee_memory_runtime_binding.py diff --git a/docs/proposals/music-studio-template.md b/docs/proposals/music-studio-template.md index 2a46c6a..85a1484 100644 --- a/docs/proposals/music-studio-template.md +++ b/docs/proposals/music-studio-template.md @@ -1,12 +1,28 @@ # Live Music Studio integrations -Status: Proposed · Last updated: 2026-09-29 +Status: In progress (private manual-audio pilot) · Last updated: 2026-10-05 The current keyless Music Studio demo uses deterministic local audio and lets an owner compare candidates, request bounded revisions, approve a result, and download a durable release package. It does **not** call Suno, search live music trends, or autonomously publish or monetize a song. The [implementation status](../implementation-status.md) describes the platform boundary. +## Current implementation boundary + +The core now binds automatic employee memory to the executing Run's pinned, verified Agent +Version and active company appointment, rather than caller-supplied employee IDs. After a +Coordinated Task completes and its configured deliverable acceptance permits delivery, each +successful winning work item can propose durable learning with its own output digest and Run +identity. These work-item proposals require human review, including when a policy normally +auto-accepts the memory type. Reprocessing does not recreate rejected or revoked proposals. + +Real production is being developed as a separate private extension, not by replacing the demo's +deterministic provider after its Task has already finished. The initial pilot uses real text +employees, manually supplied WAV candidates, actual audio-model review and bounded new rounds. +Suno API generation remains unavailable until official account access, documentation, pricing, +job reconciliation and budget semantics are verified. A text model alone cannot perform listening +review. This is not yet a qualified autonomous production or commercial-release workflow. + ## Proposed next outcome An owner configures approved model and music-generation connections, supplies a creative brief, diff --git a/docs/roadmap.md b/docs/roadmap.md index 7603970..1a2f218 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -1,6 +1,6 @@ # Roadmap -Last updated: 2026-09-30 · Status: Alpha +Last updated: 2026-10-05 · Status: Alpha The supported single-team v1 baseline and guided Console are delivered. The [implementation status](implementation-status.md) lists what actually runs, while the @@ -9,6 +9,10 @@ only; it is not a claim that a proposal has shipped. ## Next product validation +- Validate the [private real-audio Music Studio pilot](proposals/music-studio-template.md): + differentiated real-model employees, bounded revisions and evidence-bound listening review. + Employee-private memory binding and per-work-item learning proposals are implemented in core; + provider credentials, live generation and real listening qualification remain separate. - Validate [opt-in Coordinated deliverable acceptance](deliverable-acceptance.md) on repeatable real-model workflows. It distinguishes missing inputs, unit mismatches and human review without complicating simple tasks. Semantic review, automated revision and external post-acceptance diff --git a/src/agentmesh/application/memory_runtime_services.py b/src/agentmesh/application/memory_runtime_services.py index 935d937..281659b 100644 --- a/src/agentmesh/application/memory_runtime_services.py +++ b/src/agentmesh/application/memory_runtime_services.py @@ -14,7 +14,8 @@ memory_context_record, ) from agentmesh.application.ports import UnitOfWorkFactory, WorkflowWorkItem -from agentmesh.domain.company import ResourceStatus +from agentmesh.domain.company import AppointmentStatus, ResourceStatus +from agentmesh.domain.coordination import SubtaskStatus from agentmesh.domain.deliverable_acceptance import ACCEPTANCE_POLICY_INPUT_KEY from agentmesh.domain.errors import ( InvalidOrganizationalMemory, @@ -24,9 +25,11 @@ MemoryNamespaceType, MemoryPolicy, MemoryProvenanceType, + MemoryRecord, MemorySensitivity, MemoryType, ) +from agentmesh.domain.registry import AgentDefinitionLifecycle, AgentVersionStatus from agentmesh.domain.tasks import ( RunRole, RunStatus, @@ -120,6 +123,11 @@ def assemble( "format": "Return a strict JSON object with a string 'summary' and " "optional 'memory_candidates' array.", "maximum_candidates": 5, + "allowed_namespaces": [ + {"namespace_type": kind.value, "namespace_id": identifier} + for kind, identifier in namespaces + if policy.permits_namespace(kind, identifier, write=True) + ], "candidate_fields": [ "memory_type", "subject_key", @@ -131,7 +139,10 @@ def assemble( ], "guidance": ( "Propose only durable cross-Task learning. Do not include " - "credentials, raw conversation history, or unsupported claims." + "credentials, raw conversation history, or unsupported claims. " + "Use the bound EMPLOYEE namespace for personal learning; " + "use COMPANY only for genuinely shared knowledge. " + "Coordinated employee learning remains pending human review." ), } return MemoryContextAssembly( @@ -172,43 +183,88 @@ def capture_completed_task_in_unit_of_work( if not project_deliverable_acceptance(aggregate)["delivery_allowed"]: return MemoryCaptureResult(candidate_ids=(), rejected_count=0) runs = uow.runs.list_for_task(task.id) + sources: list[tuple[TaskRun, dict[str, Any], str, str, bool]] = [] + if task.execution_mode is TaskExecutionMode.COORDINATED: + runs_by_id = {value.id: value for value in runs} + for subtask in uow.subtasks.list_for_task(task.id): + winning_run = runs_by_id.get(subtask.current_run_id) + if ( + subtask.task_id == task.id + and subtask.status is SubtaskStatus.COMPLETED + and isinstance(subtask.output, dict) + and winning_run is not None + and winning_run.task_id == task.id + and winning_run.subtask_id == subtask.id + and winning_run.status is RunStatus.SUCCEEDED + and winning_run.role is RunRole.EXECUTOR + and winning_run.output == subtask.output + ): + sources.append( + (winning_run, subtask.output, "run-output", str(winning_run.id), True) + ) if task.execution_mode is TaskExecutionMode.REVIEWED: eligible = [ value for value in runs - if value.role is RunRole.EXECUTOR - and value.status is RunStatus.SUCCEEDED + if value.role is RunRole.EXECUTOR and value.status is RunStatus.SUCCEEDED ] - run = ( - max(eligible, key=lambda value: value.revision_number) - if eligible - else None - ) + run = max(eligible, key=lambda value: value.revision_number) if eligible else None else: - run = ( - uow.runs.get(task.current_run_id) - if task.current_run_id is not None - else None + run = uow.runs.get(task.current_run_id) if task.current_run_id is not None else None + if ( + run is not None + and run.task_id == task.id + and run.status is RunStatus.SUCCEEDED + and run.role is not RunRole.REVIEWER + and (task.execution_mode is not TaskExecutionMode.COORDINATED or run.subtask_id is None) + and not any(value[0].id == run.id for value in sources) + ): + sources.append((run, task.output, "task-output", str(task.id), False)) + candidate_ids: list[UUID] = [] + rejected_count = 0 + for run, output, evidence_type, evidence_id, require_review in sources: + result = self._capture_output_candidates( + uow, + task, + run, + output, + evidence_type=evidence_type, + evidence_id=evidence_id, + require_review=require_review, ) - if run is None: - return MemoryCaptureResult(candidate_ids=(), rejected_count=0) + candidate_ids.extend(result.candidate_ids) + rejected_count += result.rejected_count + return MemoryCaptureResult( + candidate_ids=tuple(dict.fromkeys(candidate_ids)), rejected_count=rejected_count + ) + + def _capture_output_candidates( + self, + uow: Any, + task: Task, + run: TaskRun, + output: dict[str, Any], + *, + evidence_type: str, + evidence_id: str, + require_review: bool, + ) -> MemoryCaptureResult: resolved = self._runtime_scope_in_unit_of_work(uow, task, run) if resolved is None: return MemoryCaptureResult(candidate_ids=(), rejected_count=0) company_id, policy, _namespaces = resolved if not policy.extraction_enabled: return MemoryCaptureResult(candidate_ids=(), rejected_count=0) - raw_candidates = task.output.get("memory_candidates", []) + raw_candidates = output.get("memory_candidates", []) if not isinstance(raw_candidates, list): return MemoryCaptureResult(candidate_ids=(), rejected_count=1) - output_digest = sha256( - json.dumps( - task.output, - sort_keys=True, - separators=(",", ":"), - default=str, - ).encode() - ).hexdigest() + try: + output_digest = sha256( + json.dumps(output, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + ).hexdigest() + except (TypeError, ValueError): + return MemoryCaptureResult(candidate_ids=(), rejected_count=len(raw_candidates)) + existing_records = uow.organizational_memory.list_records(company_id) candidate_ids: list[UUID] = [] rejected = max(0, len(raw_candidates) - 5) for raw in raw_candidates[:5]: @@ -216,39 +272,83 @@ def capture_completed_task_in_unit_of_work( rejected += 1 continue try: - snapshot = self._memory_service.propose_in_unit_of_work( - uow, - company_id, - policy_id=policy.id, - namespace_type=MemoryNamespaceType( - str(raw.get("namespace_type", "COMPANY")).upper() - ), - namespace_id=str( - raw.get("namespace_id") or company_id - ), + namespace_type = MemoryNamespaceType( + str(raw.get("namespace_type", "COMPANY")).upper() + ) + namespace_id = str(raw.get("namespace_id") or company_id) + allowed_scopes = self._namespace_candidates( + uow, task, run, company_id, task.input["company_context"] + ) + if (namespace_type, namespace_id) not in allowed_scopes: + raise InvalidOrganizationalMemory( + "Automatic candidate namespace is not bound to this Run" + ) + proposed = MemoryRecord.propose( + company_id=company_id, + namespace_type=namespace_type, + namespace_id=namespace_id, memory_type=MemoryType(str(raw["memory_type"]).upper()), content=str(raw["content"]), subject_key=raw.get("subject_key"), provenance_type=MemoryProvenanceType.TASK, provenance_id=str(task.id), confidence_basis_points=min( - int(raw.get("confidence_basis_points", 5_000)), - 7_500, + int(raw.get("confidence_basis_points", 5_000)), 7_500 ), - sensitivity=MemorySensitivity( - str(raw.get("sensitivity", "INTERNAL")).upper() + sensitivity=MemorySensitivity(str(raw.get("sensitivity", "INTERNAL")).upper()), + proposed_by_run_id=run.id, + ) + duplicate = next( + ( + value + for value in existing_records + if value.proposed_by_run_id == run.id + and value.provenance_type is MemoryProvenanceType.TASK + and value.provenance_id == str(task.id) + and value.namespace_type == proposed.namespace_type + and value.namespace_id == proposed.namespace_id + and value.memory_type == proposed.memory_type + and value.content_digest == proposed.content_digest + and value.subject_key == proposed.subject_key + and any( + evidence.evidence_type == evidence_type + and evidence.evidence_id == evidence_id + and evidence.evidence_digest == output_digest + for evidence in uow.organizational_memory.list_evidence(value.id) + ) ), + None, + ) + if duplicate is not None: + # Replays never resurrect rejected/revoked learning proposals. + candidate_ids.append(duplicate.id) + continue + snapshot = self._memory_service.propose_in_unit_of_work( + uow, + company_id, + policy_id=policy.id, + namespace_type=namespace_type, + namespace_id=namespace_id, + memory_type=proposed.memory_type, + content=proposed.content, + subject_key=proposed.subject_key, + provenance_type=MemoryProvenanceType.TASK, + provenance_id=str(task.id), + confidence_basis_points=proposed.confidence_basis_points, + sensitivity=proposed.sensitivity, evidence=[ { - "evidence_type": "task-output", - "evidence_id": str(task.id), + "evidence_type": evidence_type, + "evidence_id": evidence_id, "evidence_digest": output_digest, } ], proposed_by_run_id=run.id, actor=f"agent:{run.agent_id}", + require_review=require_review, ) candidate_ids.append(snapshot.memory.id) + existing_records.append(snapshot.memory) except ( KeyError, TypeError, @@ -269,11 +369,16 @@ def capture_completed_task_in_unit_of_work( def _runtime_scope( self, task: Task, run: TaskRun - ) -> tuple[ - UUID, - MemoryPolicy, - list[tuple[MemoryNamespaceType, str]], - ] | None: + ) -> ( + tuple[ + UUID, + MemoryPolicy, + list[tuple[MemoryNamespaceType, str]], + ] + | None + ): + if task.tenant_id != self._tenant_id or run.task_id != task.id: + return None context = task.input.get("company_context") if not isinstance(context, dict): return None @@ -285,11 +390,16 @@ def _runtime_scope_in_unit_of_work( uow: Any, task: Task, run: TaskRun, - ) -> tuple[ - UUID, - MemoryPolicy, - list[tuple[MemoryNamespaceType, str]], - ] | None: + ) -> ( + tuple[ + UUID, + MemoryPolicy, + list[tuple[MemoryNamespaceType, str]], + ] + | None + ): + if task.tenant_id != self._tenant_id or run.task_id != task.id: + return None context = task.input.get("company_context") if not isinstance(context, dict): return None @@ -300,20 +410,16 @@ def _runtime_scope_in_unit_of_work( company = uow.company_model.get_company(company_id) if company is None or company.tenant_id != self._tenant_id: return None - position = self._position_for_run(uow, company_id, context, run) + _definition_id, position = self._employee_binding(uow, company_id, context, run) policy = None requested_policy_id = context.get("memory_policy_id") if requested_policy_id: try: - policy = uow.organizational_memory.get_policy( - UUID(str(requested_policy_id)) - ) + policy = uow.organizational_memory.get_policy(UUID(str(requested_policy_id))) except ValueError: return None elif position is not None and position.memory_policy_id is not None: - policy = uow.organizational_memory.get_policy( - position.memory_policy_id - ) + policy = uow.organizational_memory.get_policy(position.memory_policy_id) else: active = [ value @@ -324,70 +430,107 @@ def _runtime_scope_in_unit_of_work( policy = active[0] if policy is None or policy.company_id != company_id or not policy.active: return None + candidates = self._namespace_candidates(uow, task, run, company_id, context) + namespaces = [ + value for value in candidates if policy.permits_namespace(*value, write=False) + ] + if not namespaces: + return None + return company_id, policy, namespaces + + def _namespace_candidates( + self, + uow: Any, + task: Task, + run: TaskRun, + company_id: UUID, + context: dict[str, Any], + ) -> list[tuple[MemoryNamespaceType, str]]: candidates = [ (MemoryNamespaceType.COMPANY, str(company_id)), (MemoryNamespaceType.PROJECT, task.project_id), ] unit_id = context.get("organization_unit_id") if unit_id: - candidates.append((MemoryNamespaceType.UNIT, str(unit_id))) + try: + unit = uow.company_model.get_unit(UUID(str(unit_id))) + except ValueError: + unit = None + if ( + unit is not None + and unit.company_id == company_id + and unit.status is ResourceStatus.ACTIVE + ): + candidates.append((MemoryNamespaceType.UNIT, str(unit.id))) + definition_id, position = self._employee_binding(uow, company_id, context, run) if position is not None: - candidates.append( - (MemoryNamespaceType.POSITION, str(position.id)) - ) - definition_id = self._agent_definition_id(context, run) + candidates.append((MemoryNamespaceType.POSITION, str(position.id))) if definition_id: - candidates.append( - (MemoryNamespaceType.EMPLOYEE, definition_id) - ) - namespaces = [ - value - for value in candidates - if policy.permits_namespace(*value, write=False) - ] - if not namespaces: - return None - return company_id, policy, namespaces + candidates.append((MemoryNamespaceType.EMPLOYEE, definition_id)) + return candidates - @staticmethod - def _position_for_run( + def _employee_binding( + self, uow: Any, company_id: UUID, context: dict[str, Any], run: TaskRun, - ) -> Any | None: - for worker in context.get("workforce", []): - if ( - isinstance(worker, dict) - and worker.get("agent_name") == run.agent_id - and worker.get("position_id") - ): - try: - position = uow.company_model.get_position( - UUID(str(worker["position_id"])) - ) - except ValueError: - return None - if ( - position - and position.company_id == company_id - and position.status is ResourceStatus.ACTIVE - ): - return position - return None - - @staticmethod - def _agent_definition_id( - context: dict[str, Any], run: TaskRun - ) -> str | None: - for worker in context.get("workforce", []): - if ( - isinstance(worker, dict) - and worker.get("agent_name") == run.agent_id - and worker.get("agent_definition_id") - ): - return str(worker["agent_definition_id"]) - return None + ) -> tuple[str | None, Any | None]: + """Derive private scopes from registry identity, never caller workforce IDs.""" + if run.role is RunRole.REVIEWER or run.agent_version_id is None: + return None, None + version = uow.agent_versions.get(run.agent_version_id) + if ( + version is None + or version.status is not AgentVersionStatus.PUBLISHED + or not version.content_digest + or version.content_digest != run.agent_version_digest + ): + return None, None + definition = uow.agent_definitions.get(version.definition_id) + if ( + definition is None + or definition.tenant_id != self._tenant_id + or definition.name != run.agent_id + or definition.lifecycle is not AgentDefinitionLifecycle.ACTIVE + ): + return None, None + appointments = [ + value + for value in uow.company_model.list_appointments(company_id) + if value.status is AppointmentStatus.ACTIVE + and value.company_id == company_id + and value.agent_definition_id == definition.id + and value.agent_version_id == version.id + ] + workforce = context.get("workforce", []) + if not isinstance(workforce, list): + return None, None + hints = [ + value + for value in workforce + if isinstance(value, dict) and value.get("agent_name") == run.agent_id + ] + if len(hints) > 1: + return None, None + if hints and hints[0].get("position_id"): + try: + position_id = UUID(str(hints[0]["position_id"])) + except ValueError: + return None, None + appointments = [value for value in appointments if value.position_id == position_id] + positions = [ + position + for appointment in appointments + if (position := uow.company_model.get_position(appointment.position_id)) is not None + and position.company_id == company_id + and position.status is ResourceStatus.ACTIVE + ] + if not positions: + return None, None + # A caller hint may select a verified appointment, but cannot invent one. + # With several appointments, employee scope is known; position scope isn't. + return str(definition.id), positions[0] if len(positions) == 1 else None @staticmethod def _base_work_item( diff --git a/src/agentmesh/application/organizational_memory_services.py b/src/agentmesh/application/organizational_memory_services.py index 4e0ba3e..46978a7 100644 --- a/src/agentmesh/application/organizational_memory_services.py +++ b/src/agentmesh/application/organizational_memory_services.py @@ -423,6 +423,7 @@ def propose_in_unit_of_work( supersedes_id: UUID | None = None, expires_at: datetime | None = None, actor: str, + require_review: bool = False, actor_roles: set[str] | None = None, subject_key: str | None = None, ) -> MemorySnapshot: @@ -485,7 +486,7 @@ def propose_in_unit_of_work( for item in evidence_records: uow.organizational_memory.add_evidence(item) reviews: list[MemoryReview] = [] - if memory_type in policy.auto_accept_memory_types: + if not require_review and memory_type in policy.auto_accept_memory_types: if supersedes_id is not None: original = self._memory(uow, company_id, supersedes_id, for_update=True) original.supersede() diff --git a/tests/test_employee_memory_runtime_binding.py b/tests/test_employee_memory_runtime_binding.py new file mode 100644 index 0000000..20a9c7e --- /dev/null +++ b/tests/test_employee_memory_runtime_binding.py @@ -0,0 +1,434 @@ +from copy import deepcopy +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +from agentmesh.application.memory_runtime_services import RuntimeMemoryService +from agentmesh.domain.coordination import Subtask, SubtaskStatus +from agentmesh.domain.deliverable_acceptance import ACCEPTANCE_POLICY_INPUT_KEY +from agentmesh.domain.organizational_memory import MemoryNamespaceType, MemoryStatus, MemoryType +from agentmesh.domain.registry import AgentVersionStatus +from agentmesh.domain.tasks import RunRole, RunStatus, TaskExecutionMode, TaskStatus +from agentmesh.features import FeatureGateSet +from tests.test_organizational_memory import _accept, _company, _policy, _propose + + +@pytest.fixture +def employee_memory( + company_service, organizational_memory_service, task_service, registry_service, uow_factory +): + company, unit = _company(company_service) + policy = _policy( + organizational_memory_service, + company.id, + readable=[f"company/{company.id}", "employee/*", "position/*", "project/*", "unit/*"], + writable=[f"company/{company.id}", "employee/*", "position/*", "project/*", "unit/*"], + extraction_enabled=True, + ) + definition = next( + value + for value in uow_factory.store.agent_definitions.values() + if value.name == "test-agent" + ) + version = uow_factory.store.agent_versions[definition.default_version_id] + other = registry_service.ensure_builtin_agent("other-employee") + position = company_service.create_position( + company.id, + primary_unit_id=unit.id, + key="lyricist", + title="Lyricist", + responsibility_contract={"outcomes": ["lyrics"]}, + memory_policy_id=policy.id, + ) + other_position = company_service.create_position( + company.id, + primary_unit_id=unit.id, + key="critic", + title="Critic", + responsibility_contract={"outcomes": ["critique"]}, + ) + appointment = company_service.appoint( + company.id, + position_id=position.id, + agent_definition_id=definition.id, + agent_version_id=version.id, + appointed_by="owner", + reason="Appoint trusted lyricist.", + ) + company_service.appoint( + company.id, + position_id=other_position.id, + agent_definition_id=other.definition.id, + agent_version_id=other.definition.default_version_id, + appointed_by="owner", + reason="Appoint independent critic.", + ) + aggregate = task_service.create_task( + "Write lyrics with company guidance", + {"company_context": {"company_id": str(company.id)}}, + ) + aggregate = task_service.request_run(aggregate.task.id) + run = aggregate.runs[0] + runtime = RuntimeMemoryService( + uow_factory=uow_factory, + memory_service=organizational_memory_service, + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", "company_model=true,organizational_memory=true" + ), + ) + return SimpleNamespace( + company=company, + unit=unit, + policy=policy, + definition=definition, + version=version, + other=other.definition, + position=position, + other_position=other_position, + appointment=appointment, + task=aggregate.task, + run=run, + runtime=runtime, + memory=organizational_memory_service, + company_service=company_service, + uow_factory=uow_factory, + ) + + +def _scopes(value): + result = value.runtime._runtime_scope(value.task, value.run) + assert result is not None + return result[2] + + +def test_employee_and_position_are_bound_without_caller_workforce(employee_memory): + value = employee_memory + scopes = _scopes(value) + assert (MemoryNamespaceType.EMPLOYEE, str(value.definition.id)) in scopes + assert (MemoryNamespaceType.POSITION, str(value.position.id)) in scopes + assert (MemoryNamespaceType.EMPLOYEE, str(value.other.id)) not in scopes + + +def test_caller_cannot_spoof_employee_definition(employee_memory): + value = employee_memory + value.task.input["company_context"]["workforce"] = [ + {"agent_name": value.run.agent_id, "agent_definition_id": str(value.other.id)} + ] + scopes = _scopes(value) + assert (MemoryNamespaceType.EMPLOYEE, str(value.definition.id)) in scopes + assert (MemoryNamespaceType.EMPLOYEE, str(value.other.id)) not in scopes + + +def test_caller_cannot_select_another_employees_position(employee_memory): + value = employee_memory + value.task.input["company_context"]["workforce"] = [ + {"agent_name": value.run.agent_id, "position_id": str(value.other_position.id)} + ] + scopes = _scopes(value) + assert not any( + kind in {MemoryNamespaceType.EMPLOYEE, MemoryNamespaceType.POSITION} for kind, _ in scopes + ) + assert (MemoryNamespaceType.COMPANY, str(value.company.id)) in scopes + + +@pytest.mark.parametrize("invalid", ["unknown_version", "wrong_digest", "revoked", "wrong_name"]) +def test_invalid_pinned_identity_never_opens_private_scope(employee_memory, invalid): + value = employee_memory + if invalid == "unknown_version": + value.run.agent_version_id = uuid4() + elif invalid == "wrong_digest": + value.run.agent_version_digest = "0" * 64 + elif invalid == "revoked": + value.uow_factory.store.agent_versions[value.version.id].status = AgentVersionStatus.REVOKED + else: + value.run.agent_id = value.other.name + assert not any( + kind in {MemoryNamespaceType.EMPLOYEE, MemoryNamespaceType.POSITION} + for kind, _ in _scopes(value) + ) + + +def test_ended_appointment_revokes_employee_and_position_scope(employee_memory): + value = employee_memory + value.company_service.end_appointment(value.company.id, value.appointment.id) + assert not any( + kind in {MemoryNamespaceType.EMPLOYEE, MemoryNamespaceType.POSITION} + for kind, _ in _scopes(value) + ) + + +def test_registry_default_change_does_not_change_pinned_employee_scope(employee_memory): + value = employee_memory + value.uow_factory.store.agent_definitions[value.definition.id].default_version_id = uuid4() + assert (MemoryNamespaceType.EMPLOYEE, str(value.definition.id)) in _scopes(value) + + +def test_reading_employee_memory_also_keeps_shared_company_knowledge(employee_memory): + value = employee_memory + own = _propose( + value.memory, + value.company.id, + value.policy.id, + "Write lyrics with concise imagery.", + namespace_type=MemoryNamespaceType.EMPLOYEE, + namespace_id=str(value.definition.id), + ) + other = _propose( + value.memory, + value.company.id, + value.policy.id, + "Write lyrics with private critique.", + namespace_type=MemoryNamespaceType.EMPLOYEE, + namespace_id=str(value.other.id), + ) + shared = _propose( + value.memory, value.company.id, value.policy.id, "Company guidance: avoid imitation." + ) + for memory in (own, other, shared): + _accept(value.memory, value.company.id, value.policy.id, memory.memory.id) + assembly = value.runtime.assemble(value.task, value.run, None) + ids = {item["memory_id"] for item in assembly.work_item.input["agentmesh_memory"]["records"]} + assert str(own.memory.id) in ids + assert str(shared.memory.id) in ids + assert str(other.memory.id) not in ids + + +@pytest.mark.parametrize("namespace", ["EMPLOYEE", "POSITION", "UNIT", "USER"]) +def test_wildcard_policy_cannot_write_unbound_automatic_candidate(employee_memory, namespace): + value = employee_memory + value.task.status = TaskStatus.COMPLETED + value.task.current_run_id = value.run.id + value.task.output = { + "summary": "Completed lyrics", + "memory_candidates": [ + { + "memory_type": "PATTERN", + "content": "Untrusted private learning.", + "namespace_type": namespace, + "namespace_id": str(uuid4()), + } + ], + } + value.run.status = RunStatus.SUCCEEDED + value.uow_factory.store.runs[value.run.id] = deepcopy(value.run) + with value.uow_factory() as uow: + result = value.runtime.capture_completed_task_in_unit_of_work(uow, value.task) + uow.commit() + assert result.candidate_ids == () + assert result.rejected_count == 1 + + +def test_trusted_employee_candidate_is_governed_and_attributed(employee_memory): + value = employee_memory + value.task.status = TaskStatus.COMPLETED + value.task.current_run_id = value.run.id + value.task.output = { + "summary": "Completed lyrics", + "memory_candidates": [ + { + "memory_type": "PATTERN", + "content": "Keep lyrics concise.", + "namespace_type": "EMPLOYEE", + "namespace_id": str(value.definition.id), + } + ], + } + value.run.status = RunStatus.SUCCEEDED + value.uow_factory.store.runs[value.run.id] = deepcopy(value.run) + with value.uow_factory() as uow: + result = value.runtime.capture_completed_task_in_unit_of_work(uow, value.task) + uow.commit() + assert len(result.candidate_ids) == 1 + memory = value.memory.list_candidates(value.company.id)[0].memory + assert memory.namespace_id == str(value.definition.id) + assert memory.proposed_by_run_id == value.run.id + + +def test_reviewer_is_not_automatically_given_memory(employee_memory): + value = employee_memory + value.run.role = RunRole.REVIEWER + assert value.runtime.assemble(value.task, value.run, None).search is None + + +def test_wrong_task_binding_or_tenant_cannot_assemble_memory(employee_memory): + value = employee_memory + value.run.task_id = uuid4() + assert value.runtime.assemble(value.task, value.run, None).search is None + + +def _completed_work_item(value, *, other=False): + run = deepcopy(value.run) + run.id = uuid4() + run.subtask_id = uuid4() + run.status = RunStatus.SUCCEEDED + definition = value.other if other else value.definition + version = value.uow_factory.store.agent_versions[definition.default_version_id] + run.agent_id = definition.name + run.agent_version_id = version.id + run.agent_version_digest = version.content_digest + run.output = { + "summary": "Completed work item", + "memory_candidates": [ + { + "memory_type": "PATTERN", + "content": "Critic checks audio evidence." + if other + else "Lyricist uses clear imagery.", + "namespace_type": "EMPLOYEE", + "namespace_id": str(definition.id), + } + ], + } + subtask = Subtask.create( + subtask_id=run.subtask_id, + task_id=value.task.id, + key="critique" if other else "lyrics", + objective="Complete isolated employee work", + input={}, + required_capabilities=(), + preferred_agent_id=definition.name, + initially_ready=True, + ) + subtask.status = SubtaskStatus.COMPLETED + subtask.current_run_id = run.id + subtask.output = deepcopy(run.output) + value.uow_factory.store.runs[run.id] = run + value.uow_factory.store.subtasks[subtask.id] = subtask + value.task.status = TaskStatus.COMPLETED + value.task.execution_mode = TaskExecutionMode.COORDINATED + value.task.current_run_id = None + value.task.output = {"summary": "Company completed music task"} + return run, subtask + + +def _capture(value): + with value.uow_factory() as uow: + result = value.runtime.capture_completed_task_in_unit_of_work(uow, value.task) + uow.commit() + return result + + +def test_completed_coordinated_task_collects_each_employees_own_learning(employee_memory): + value = employee_memory + lyricist, _ = _completed_work_item(value) + critic, _ = _completed_work_item(value, other=True) + result = _capture(value) + assert len(result.candidate_ids) == 2 + candidates = value.memory.list_candidates(value.company.id) + assert {item.memory.proposed_by_run_id for item in candidates} == {lyricist.id, critic.id} + assert {item.memory.namespace_id for item in candidates} == { + str(value.definition.id), + str(value.other.id), + } + for item in candidates: + assert item.memory.status is MemoryStatus.CANDIDATE + assert item.evidence[0].evidence_type == "run-output" + assert item.evidence[0].evidence_id == str(item.memory.proposed_by_run_id) + assert len(item.evidence[0].evidence_digest) == 64 + + +@pytest.mark.parametrize("overall", [TaskStatus.RUNNING, TaskStatus.FAILED, TaskStatus.CANCELED]) +def test_employee_learning_waits_for_overall_task_completion(employee_memory, overall): + value = employee_memory + _completed_work_item(value) + value.task.status = overall + assert _capture(value).candidate_ids == () + assert value.memory.list_candidates(value.company.id) == [] + + +@pytest.mark.parametrize("bad", ["failed_run", "wrong_task", "wrong_subtask", "wrong_output"]) +def test_only_matching_successful_winning_run_is_learning_source(employee_memory, bad): + value = employee_memory + run, subtask = _completed_work_item(value) + if bad == "failed_run": + run.status = RunStatus.FAILED + elif bad == "wrong_task": + run.task_id = uuid4() + elif bad == "wrong_subtask": + run.subtask_id = uuid4() + else: + subtask.output["summary"] = "Output does not match winning Run" + value.task.current_run_id = run.id + value.task.output = deepcopy(run.output) + assert _capture(value).candidate_ids == () + + +def test_superseded_successful_retry_does_not_propose_learning(employee_memory): + value = employee_memory + winning, _ = _completed_work_item(value) + discarded = deepcopy(winning) + discarded.id = uuid4() + discarded.output["memory_candidates"][0]["content"] = "Discarded attempt must not be learned." + value.uow_factory.store.runs[discarded.id] = discarded + assert len(_capture(value).candidate_ids) == 1 + assert value.memory.list_candidates(value.company.id)[0].memory.proposed_by_run_id == winning.id + + +def test_subtask_learning_requires_review_even_with_auto_accept_policy(employee_memory): + value = employee_memory + _completed_work_item(value) + value.uow_factory.store.memory_policies[value.policy.id].auto_accept_memory_types = [ + MemoryType.PATTERN + ] + assert len(_capture(value).candidate_ids) == 1 + memory = value.memory.list_candidates(value.company.id)[0].memory + assert memory.status is MemoryStatus.CANDIDATE + assert memory.reviewed_by is None + + +def test_completion_replay_does_not_recreate_rejected_personal_learning(employee_memory): + value = employee_memory + _completed_work_item(value) + first = _capture(value) + value.memory.review( + value.company.id, + first.candidate_ids[0], + policy_id=value.policy.id, + decision="REJECT", + reviewer="owner", + reviewer_roles={"TENANT_ADMIN"}, + reason="This learning is not reliable.", + ) + second = _capture(value) + assert second.candidate_ids == first.candidate_ids + assert second.rejected_count == 0 + with value.uow_factory() as uow: + records = uow.organizational_memory.list_records(value.company.id) + assert len(records) == 1 + assert records[0].status is MemoryStatus.REJECTED + + +def test_coordinated_final_run_does_not_duplicate_subtask_learning(employee_memory): + value = employee_memory + run, _ = _completed_work_item(value) + value.task.current_run_id = run.id + value.task.output = deepcopy(run.output) + assert len(_capture(value).candidate_ids) == 1 + + +def test_coordinated_employee_learning_waits_for_deliverable_acceptance(employee_memory): + value = employee_memory + _completed_work_item(value) + value.task.input[ACCEPTANCE_POLICY_INPUT_KEY] = { + "version": 1, + "target_subtask_key": "lyrics", + "require_human_review": True, + "checks": [ + { + "key": "summary", + "description": "Summary must exist", + "kind": "OUTPUT_PATH_EXISTS", + "path": ["summary"], + "required": True, + } + ], + } + assert _capture(value).candidate_ids == () + assert value.memory.list_candidates(value.company.id) == [] + value.task.input[ACCEPTANCE_POLICY_INPUT_KEY]["require_human_review"] = False + assert len(_capture(value).candidate_ids) == 1 + value.run.task_id = value.task.id + value.task.tenant_id = "another-tenant" + assert value.runtime.assemble(value.task, value.run, None).search is None From a0109d1ebd3dc0dbca734b0a4ae433fad8de7dc0 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Mon, 5 Oct 2026 16:18:03 +0800 Subject: [PATCH 2/2] feat: preserve business fields declared by pinned employee output schema --- src/agentmesh/orchestration/model_agent.py | 15 +++++++- tests/test_model_pinned_output_contract.py | 43 ++++++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) create mode 100644 tests/test_model_pinned_output_contract.py diff --git a/src/agentmesh/orchestration/model_agent.py b/src/agentmesh/orchestration/model_agent.py index b6ce2f6..12d82af 100644 --- a/src/agentmesh/orchestration/model_agent.py +++ b/src/agentmesh/orchestration/model_agent.py @@ -344,7 +344,8 @@ def execute_version( ) text = self._output_text(response) structured = self._structured_result( - text, acceptance_contract=input.get("agentmesh_deliverable_contract") + text, acceptance_contract=input.get("agentmesh_deliverable_contract"), + output_schema=version.output_schema, ) return { **structured, @@ -371,6 +372,7 @@ def execute_version( @staticmethod def _structured_result( text: str, acceptance_contract: dict[str, Any] | None = None, + output_schema: dict[str, Any] | None = None, ) -> dict[str, Any]: """Preserve normal text and explicitly governed structured output contracts.""" try: @@ -380,6 +382,17 @@ def _structured_result( if not isinstance(value, dict) or not isinstance(value.get("summary"), str): return {"summary": text} result: dict[str, Any] = {"summary": value["summary"]} + # Only the immutable registry contract can declare general business roots. + # Task input and the model itself cannot expand the platform output envelope. + properties = output_schema.get("properties") if isinstance(output_schema, dict) else None + if isinstance(properties, dict): + for root in list(properties)[:60]: + if ( + isinstance(root, str) + and root not in {"agent", "execution", "memory_candidates"} + and root in value + ): + result[root] = value[root] if ( isinstance(acceptance_contract, dict) and type(acceptance_contract.get("version")) is int diff --git a/tests/test_model_pinned_output_contract.py b/tests/test_model_pinned_output_contract.py new file mode 100644 index 0000000..f7dfe73 --- /dev/null +++ b/tests/test_model_pinned_output_contract.py @@ -0,0 +1,43 @@ +import json + +from agentmesh.orchestration.model_agent import OpenAIResponsesAgentExecutor + + +def test_pinned_employee_schema_preserves_business_roots(): + result = OpenAIResponsesAgentExecutor._structured_result( + json.dumps({"summary": "Draft", "lyrics": "Original words", "ignored": "not declared"}), + output_schema={"type": "object", "properties": {"lyrics": {"type": "string"}}}, + ) + assert result == {"summary": "Draft", "lyrics": "Original words"} + + +def test_pinned_schema_cannot_replace_platform_identity_or_execution(): + result = OpenAIResponsesAgentExecutor._structured_result( + json.dumps({"summary": "Plan", "agent": "spoof", "execution": "spoof", + "composition": {"tempo": 100}}), + output_schema={"properties": {"agent": {}, "execution": {}, "composition": {}}}, + ) + assert result == {"summary": "Plan", "composition": {"tempo": 100}} + + +def test_default_contract_does_not_keep_arbitrary_model_fields(): + result = OpenAIResponsesAgentExecutor._structured_result( + '{"summary":"ok","lyrics":"undeclared"}', output_schema={"type": "object"}, + ) + assert result == {"summary": "ok"} + + +def test_existing_acceptance_roots_and_pinned_schema_can_coexist(): + result = OpenAIResponsesAgentExecutor._structured_result( + '{"summary":"ok","claim":9,"lyrics":"original"}', + acceptance_contract={"version": 1, "output_roots": ["claim"]}, + output_schema={"properties": {"lyrics": {}}}, + ) + assert result == {"summary": "ok", "claim": 9, "lyrics": "original"} + + +def test_malformed_json_remains_prose_without_invented_fields(): + result = OpenAIResponsesAgentExecutor._structured_result( + "unfinished {", output_schema={"properties": {"lyrics": {}}}, + ) + assert result == {"summary": "unfinished {"}