import copy
import json
from types import SimpleNamespace

import pytest
from fastapi import HTTPException
from sqlmodel import Session, SQLModel, create_engine, select
from app.main import segment_edit, get_campaign, segment_versions, resume_campaign_run, cancel_campaign_run
from app.models import Agent, AppUser, Campaign, CampaignRun, Canvas, CanvasGrant, Segment, Memory, ModelProvider
from app.segment_assignment import available_agents
from app.agents.snapshot import freeze_agent_snapshot
from app.agents import runner
from app.seed import upstream_anchor_cards

@pytest.fixture
def scope(tmp_path):
    engine=create_engine('sqlite://')
    SQLModel.metadata.create_all(engine)
    with Session(engine) as db:
        user=AppUser(username='owner',role='admin')
        other=AppUser(username='other',role='editor')
        db.add_all([user,other]);db.commit()
        a=Agent(name='心法智能体',role='production',rules='OLD_HEART_RULES',content_tags=['心法'],kb_ids=[1],owner_user_id=user.id)
        b=Agent(name='功法智能体',role='production',rules='NEW_PRACTICE_RULES',content_tags=['功法'],kb_ids=[2],owner_user_id=user.id)
        db.add_all([a,b]);db.commit()
        campaign=Campaign(name='循序训练',owner_user_id=user.id,status='parsed',qc_mode='off',mindmap_text='基础认知→练习→应用',canvas_snapshot={'nodes':[{'node_id':'n1','agent_id':a.id}], 'edges':[]})
        db.add(campaign);db.commit()
        segment=Segment(campaign_id=campaign.id,agent_id=a.id,node_id='n1',title='基础练习',content_tag='心法',mindmap_node='依照已学概念练习动作')
        db.add(segment);db.commit()
        yield db,user,other,a,b,campaign,segment,tmp_path
    engine.dispose()

def request(user):return SimpleNamespace(state=SimpleNamespace(user=user))

def test_manual_assignment_refreshes_configuration_preserves_task_and_sticks_after_tag_edit(scope):
    db,user,other,a,b,c,seg,_=scope
    original=copy.deepcopy(c.canvas_snapshot)
    seg.runtime_snapshot={'primary':{'id':a.id,'rules':'frozen old','kb_ids':[1]}}
    seg.kb_category_ids=[10];seg.kb_category_recommendation={'category_ids':[10]};seg.fusion_agent_ids=[a.id]
    db.add(seg);db.commit()
    segment_edit(c.id,seg.id,{'agent_id':b.id},db,request(user))
    assert seg.agent_id==b.id and seg.agent_assignment_mode=='manual'
    assert seg.runtime_snapshot=={} and seg.kb_category_ids==[] and seg.kb_category_recommendation=={}
    assert seg.fusion_agent_ids==[] and seg.node_id=='n1' and seg.mindmap_node=='依照已学概念练习动作'
    assert c.canvas_snapshot==original
    segment_edit(c.id,seg.id,{'content_tag':'心法进阶'},db,request(user))
    assert seg.agent_id==b.id
    detail=get_campaign(c.id,db,request(user))
    assert detail['segments'][0]['agent_name']=='功法智能体'
    assert detail['segments'][0]['agent_assignment_mode']=='manual' and detail['can_edit_agents']

@pytest.mark.parametrize('bad_id',[None,'abc',0,-2])
def test_invalid_agent_id_never_silently_succeeds(scope,bad_id):
    db,user,_,a,b,c,seg,_=scope
    with pytest.raises(HTTPException) as exc:segment_edit(c.id,seg.id,{'agent_id':bad_id},db,request(user))
    assert exc.value.status_code==422 and seg.agent_id==a.id


def test_foreign_disabled_and_nonproduction_agents_are_rejected(scope):
    db,user,_,a,b,c,seg,_=scope
    for item in [Agent(name='foreign',role='production',organization_id=2,visibility='global'),
                 Agent(name='disabled',role='production',status='disabled'),Agent(name='QC',role='qc')]:
        db.add(item);db.commit()
        with pytest.raises(HTTPException) as exc:segment_edit(c.id,seg.id,{'agent_id':item.id},db,request(user))
        assert exc.value.status_code==404
    assert seg.agent_id==a.id


def test_selection_obeys_canvas_use_grant_without_exposing_other_private_agents(scope):
    db,user,viewer,a,b,c,seg,_=scope
    c.owner_user_id=viewer.id
    a.visibility = b.visibility = 'private'
    db.add_all([a, b])
    canvas=Canvas(name='授权画布',owner_user_id=user.id,visibility='shared',nodes=[{'agent_id':b.id}])
    db.add_all([c,canvas]);db.commit()
    assert available_agents(db,c,viewer)==[]
    db.add(CanvasGrant(canvas_id=canvas.id,user_id=viewer.id,permission='use'));db.commit()
    assert [item.id for item in available_agents(db,c,viewer)]==[b.id]
    segment_edit(c.id,seg.id,{'agent_id':b.id},db,request(viewer))
    assert seg.agent_id==b.id


def test_existing_outputs_require_confirmation_archive_history_and_invalidate_later_drafts(scope):
    db,user,_,a,b,c,seg,path=scope
    old=path/'old.md';old.write_text('原心法正文',encoding='utf8')
    later=path/'later.md';later.write_text('依赖旧心法的下文',encoding='utf8')
    seg.status='approved';seg.artifact_path=str(old);seg.anchor_card={'ending_hook':'旧钩子'}
    next_seg=Segment(campaign_id=c.id,agent_id=a.id,title='下游应用',status='approved',artifact_path=str(later),anchor_card={'ending_hook':'旧应用'})
    skipped=Segment(campaign_id=c.id,agent_id=a.id,title='已放弃',status='skipped',artifact_path=str(later))
    run=CampaignRun(campaign_id=c.id,status='completed',config_snapshot={'original':'immutable'})
    db.add_all([seg,next_seg,skipped,run]);db.commit()
    with pytest.raises(HTTPException) as exc:segment_edit(c.id,seg.id,{'agent_id':b.id},db,request(user))
    assert exc.value.status_code==409 and seg.agent_id==a.id and next_seg.status=='approved'
    result=segment_edit(c.id,seg.id,{'agent_id':b.id,'reset_generated':True},db,request(user))
    assert result['reset_segment_ids']==[seg.id,next_seg.id]
    assert seg.status==next_seg.status=='pending' and skipped.status=='skipped'
    assert seg.artifact_path==next_seg.artifact_path=='' and seg.anchor_card==next_seg.anchor_card=={}
    assert old.read_text()=='原心法正文' and later.read_text()=='依赖旧心法的下文'
    assert seg.round==2 and next_seg.round==2
    assert segment_versions(c.id,seg.id,db,request(user))[0]['artifact']=='原心法正文'
    assert run.config_snapshot=={'original':'immutable'}
    with pytest.raises(HTTPException) as exc:resume_campaign_run(run.id,db,request(user))
    assert exc.value.status_code==409
    assert runner.human_review(db,seg.id,'approve')['status']=='error'

@pytest.mark.parametrize('state',['running','paused'])
def test_active_run_blocks_switch_until_stopped(scope,state):
    db,user,_,a,b,c,seg,_=scope
    run=CampaignRun(campaign_id=c.id,status=state)
    db.add(run);db.commit()
    with pytest.raises(HTTPException) as exc:segment_edit(c.id,seg.id,{'agent_id':b.id},db,request(user))
    assert exc.value.status_code==409 and seg.agent_id==a.id
    if state=='paused':
        assert cancel_campaign_run(run.id,db,request(user))['status']=='cancelled'
        segment_edit(c.id,seg.id,{'agent_id':b.id},db,request(user))
        assert seg.agent_id==b.id


def test_switch_uses_new_agent_rules_memory_and_kb_with_same_previous_and_next_context(scope,monkeypatch):
    db,user,_,a,b,c,seg,path=scope
    previous=path/'previous.md';previous.write_text('上一段已经确认：先分辨感受，再进行具体练习。',encoding='utf8')
    seg.title='前置概念';seg.status='approved';seg.artifact_path=str(previous);seg.anchor_card={'ending_hook':'接下来进入练习'}
    target=Segment(campaign_id=c.id,agent_id=a.id,title='基础练习',content_tag='心法',mindmap_node='运用前置概念做练习')
    ignored=Segment(campaign_id=c.id,agent_id=a.id,title='不再采用的旁支',status='skipped',node_id='skip',anchor_card={'ending_hook':'不能采用'})
    following=Segment(campaign_id=c.id,agent_id=a.id,title='成果应用',mindmap_node='运用练习的成果')
    provider=ModelProvider(name='mock',base_url='https://example.invalid/v1',models=['mock-model'],default_model='mock-model')
    db.add_all([seg,target,ignored,following,provider]);db.commit()
    db.add_all([Memory(agent_id=a.id,content='OLD_HEART_MEMORY',owner_user_id=user.id),Memory(agent_id=b.id,content='NEW_PRACTICE_MEMORY',owner_user_id=user.id)])
    db.commit()
    target.runtime_snapshot={'primary':freeze_agent_snapshot(db,a,user=user)};db.add(target);db.commit()
    segment_edit(c.id,target.id,{'agent_id':b.id},db,request(user))
    monkeypatch.setattr(runner,'ARTIFACTS_DIR',path)
    monkeypatch.setattr(runner,'_provider_model',lambda session,agent:(provider,'mock-model'))
    monkeypatch.setattr(runner,'consume_reserved_chunks',lambda *args:None)
    observed={}
    def rag(session,agent,segment,kb_ids=None):
        observed['kb_ids']=kb_ids
        return '新智能体的知识材料',[],[{'id':'stub','kb_id':2}]
    monkeypatch.setattr(runner,'_rag_materials',rag)
    def model(session,**kwargs):
        observed.update(kwargs)
        anchor={field:'明确内容' for field in runner.ANCHOR_FIELDS}
        return {'status':'completed','text':'承接前面确认的概念，进入练习。'+runner.ANCHOR_MARK+json.dumps(anchor,ensure_ascii=False)}
    monkeypatch.setattr(runner,'run_agent',model)
    result=runner.execute_segment(db,target,c)
    prompt=observed['messages'][0]['content']
    assert result['status']=='human_review' and observed['agent'].id==b.id
    assert observed['kb_ids']==[2] and 'NEW_PRACTICE_RULES' in prompt and 'NEW_PRACTICE_MEMORY' in prompt
    assert 'OLD_HEART_RULES' not in prompt and 'OLD_HEART_MEMORY' not in prompt
    assert '上一段已经确认：先分辨感受' in prompt and '下一段标题：成果应用' in prompt
    assert '标签只用于描述本段内容' in prompt and '不重复开场' in prompt
    assert runner._next_segment_context(db,target,c).find('不再采用的旁支')==-1
    assert upstream_anchor_cards(db,c.id,'end',[{'from':'skip','to':'end'}])==[]